qqtang/sskj
1
2
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
sskj/platforms/patches/kunlun_p800/hf_transformers_utils.py.patched
Quantong Qiu 227ef603a6 refactor: experiments/ + platforms/ layout and P800 SGLang support 
- Restructure repo around experiments/<name>/ and platforms/<chip>.env.
- Add shared scripts under scripts/common/ for platform/server/bench-client logic.
- Add Kunlun P800 platform config and runtime patches.
- Add dsv4_p800_sglang experiment with INT8 smoke-test support.
- Update BENCHMARK_WORKFLOW.md and README.md with chip/engine recording rules.
- Add scripts/analysis/compare_experiments.py for cross-experiment comparison.
- Ignore experiments/*/results/ raw output directories by default.
2026-07-08 05:15:59 +00:00

1452 lines
54 KiB
Plaintext
Raw Blame History

# Copyright 2023-2024 SGLang Team
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================
"""Utilities for Huggingface Transformers."""
import contextlib
import json
import logging
import os
import tempfile
import warnings
from functools import lru_cache
from pathlib import Path
from typing import Any, Dict, List, Literal, Optional, Type, Union
import torch
from huggingface_hub import snapshot_download
from sglang.srt.environ import envs
from sglang.srt.utils import get_bool_env_var
from sglang.srt.utils.runai_utils import ObjectStorageModel, is_runai_obj_uri
# Compatibility shim: flash-attn-4 registers a bare ``flash_attn`` namespace
# that makes ``is_flash_attn_2_available()`` return True, but lacks the v2 API
# (``flash_attn_func``, etc.). HuggingFace remote model code (e.g. Kimi-VL)
# guarded by that check will crash with ImportError at module load time.
# Force it to False when the real v2 API is absent.
try:
import flash_attn as _flash_attn_mod
if not hasattr(_flash_attn_mod, "flash_attn_func"):
import transformers.utils as _hf_utils
import transformers.utils.import_utils as _hf_import_utils
_hf_import_utils.is_flash_attn_2_available = lambda: False
_hf_utils.is_flash_attn_2_available = lambda: False
del _flash_attn_mod
except ImportError:
pass
# Conditional import based on SGLANG_USE_MODELSCOPE environment variable
if get_bool_env_var("SGLANG_USE_MODELSCOPE"):
from modelscope import AutoConfig, GenerationConfig
else:
from transformers import AutoConfig, GenerationConfig
from transformers import (
AutoProcessor,
AutoTokenizer,
PretrainedConfig,
PreTrainedTokenizer,
PreTrainedTokenizerBase,
PreTrainedTokenizerFast,
)
from transformers.models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from sglang.srt.configs import (
AfmoeConfig,
BailingHybridConfig,
ChatGLMConfig,
DbrxConfig,
DeepseekVL2Config,
DotsOCRConfig,
DotsVLMConfig,
ExaoneConfig,
FalconH1Config,
GraniteMoeHybridConfig,
JetNemotronConfig,
JetVLMConfig,
KimiK25Config,
KimiLinearConfig,
KimiVLConfig,
LongcatFlashConfig,
MultiModalityConfig,
NemotronH_Nano_VL_V2_Config,
NemotronHConfig,
Olmo3Config,
Qwen3_5Config,
Qwen3_5MoeConfig,
Qwen3NextConfig,
Step3p5Config,
Step3VLConfig,
)
from sglang.srt.configs.deepseek_ocr import DeepseekVLV2Config
from sglang.srt.configs.internvl import InternVLChatConfig
from sglang.srt.connector import create_remote_connector
from sglang.srt.multimodal.customized_mm_processor_utils import _CUSTOMIZED_MM_PROCESSOR
from sglang.srt.utils import is_remote_url, logger, lru_cache_frozenset, mistral_utils
_CONFIG_REGISTRY: List[Type[PretrainedConfig]] = [
AfmoeConfig,
BailingHybridConfig,
ChatGLMConfig,
DbrxConfig,
ExaoneConfig,
DeepseekVL2Config,
MultiModalityConfig,
KimiVLConfig,
InternVLChatConfig,
Step3VLConfig,
LongcatFlashConfig,
Olmo3Config,
KimiLinearConfig,
Qwen3NextConfig,
FalconH1Config,
GraniteMoeHybridConfig,
DotsVLMConfig,
DotsOCRConfig,
NemotronH_Nano_VL_V2_Config,
NemotronHConfig,
DeepseekVLV2Config,
Qwen3_5Config,
Qwen3_5MoeConfig,
JetNemotronConfig,
JetVLMConfig,
KimiK25Config,
Step3p5Config,
]
_CONFIG_REGISTRY = {
config_cls.model_type: config_cls for config_cls in _CONFIG_REGISTRY
}
for name, cls in _CONFIG_REGISTRY.items():
with contextlib.suppress(ValueError):
AutoConfig.register(name, cls)
def download_from_hf(
model_path: str,
allow_patterns: Optional[Union[str, list]] = None,
):
if os.path.exists(model_path):
return model_path
if not allow_patterns:
allow_patterns = ["*.json", "*.bin", "*.model"]
return snapshot_download(model_path, allow_patterns=allow_patterns)
def get_rope_config(config):
"""Get (rope_theta, rope_scaling) from config, supporting both v4 and v5.
In transformers v5, rope_theta/rope_scaling are accessed via the computed
property config.rope_parameters. Trust-remote-code configs or parent configs
passed to sub-models may not have this property or may return None.
Falls back to the v4-style config.rope_theta / config.rope_scaling attributes.
"""
rope_params = getattr(config, "rope_parameters", None)
if rope_params is not None and "rope_theta" in rope_params:
return rope_params["rope_theta"], rope_params
rope_theta = getattr(config, "rope_theta", None)
if rope_theta is None and rope_params is not None:
rope_theta = rope_params.get("rope_theta", None)
return rope_theta, getattr(config, "rope_scaling", None)
def _patch_text_config(parent_config: PretrainedConfig, text_config):
"""Synchronize standard attributes between parent config and text sub-config.
In transformers v5, the "untangle config" refactor removed automatic
inheritance of top-level PretrainedConfig attributes (pad_token_id,
tie_word_embeddings, etc.) from sub-configs. Downstream code expects
these attributes to be present on both configs (some models pass the
parent directly to the language model, others pass the text sub-config),
so we propagate in both directions when an attribute is missing.
(See https://github.com/huggingface/transformers/pull/41541)
"""
_ATTRS_TO_PROPAGATE = [
"pad_token_id",
"bos_token_id",
"eos_token_id",
"tie_word_embeddings",
]
for attr in _ATTRS_TO_PROPAGATE:
parent_has = hasattr(parent_config, attr)
text_has = hasattr(text_config, attr)
if parent_has and not text_has:
setattr(text_config, attr, getattr(parent_config, attr))
elif text_has and not parent_has:
setattr(parent_config, attr, getattr(text_config, attr))
return text_config
def get_hf_text_config(config: PretrainedConfig):
"""Get the "sub" config relevant to llm for multi modal models.
No op for pure text models.
"""
if config.architectures is not None:
class_name = config.architectures[0]
if class_name.startswith("Llava") and class_name.endswith("ForCausalLM"):
# We support non-hf version of llava models, so we do not want to
# read the wrong values from the unused default text_config.
# NOTE(HandH1998): We set `torch_dtype` of config to `torch.float16` for the weights, as
# `torch.float16` is default used for image features in `python/sglang/srt/models/llava.py`.
setattr(config, "dtype", torch.float16)
return config
text_config = None
# Some models (e.g. DeepSeek-OCR) store sub-configs as plain dicts.
# Convert to PretrainedConfig early so hasattr() checks and asserts work.
for _attr in ("text_config", "llm_config", "language_config", "thinker_config"):
_sub = getattr(config, _attr, None)
if isinstance(_sub, dict):
_converted = PretrainedConfig(**_sub)
# Propagate torch_dtype from parent so weight loading uses correct precision.
if (
getattr(_converted, "torch_dtype", None) is None
and getattr(config, "torch_dtype", None) is not None
):
_converted.torch_dtype = config.torch_dtype
setattr(config, _attr, _converted)
# Priority: thinker_config > llm_config > language_config > text_config
if hasattr(config, "thinker_config"):
# qwen2.5 omni
thinker_config = config.thinker_config
if hasattr(thinker_config, "text_config"):
setattr(
thinker_config.text_config,
"torch_dtype",
getattr(thinker_config, "torch_dtype", None),
)
text_config = thinker_config.text_config
else:
text_config = thinker_config
elif hasattr(config, "llm_config"):
# PointsV1.5 Chat Model
assert hasattr(config.llm_config, "num_attention_heads")
text_config = config.llm_config
elif hasattr(config, "language_config"):
text_config = config.language_config
elif hasattr(config, "text_config"):
# The code operates under the assumption that text_config should have
# `num_attention_heads` (among others). Assert here to fail early
# if transformers config doesn't align with this assumption.
assert hasattr(config.text_config, "num_attention_heads")
text_config = config.text_config
# Ensure rope_scaling dicts have "type" for remote-code compat (v5).
normalize_rope_scaling_compat(config)
if text_config is not None:
return _patch_text_config(config, text_config)
return config
# Temporary hack for DeepSeek-V3.2 model
def _load_deepseek_temp_model(
model_path: str,
model_type: Literal["deepseek_v32", "deepseek_ref"],
architecture: Literal["DeepseekV3ForCausalLM", "DeepseekV4ForCausalLM"],
trust_remote_code: bool = False,
revision: Optional[str] = None,
**kwargs,
):
# first get the local path
local_path = download_from_hf(model_path)
# then load the config file in json
backup_mode = envs.SGLANG_APPLY_CONFIG_BACKUP.get()
if backup_mode == "auto":
real_config_file = os.path.join(local_path, "config.json")
if not os.path.exists(real_config_file):
raise RuntimeError(
f"SGLANG_APPLY_CONFIG_BACKUP=auto requires the checkpoint's "
f"config.json at {real_config_file} to read num_hidden_layers."
)
with open(real_config_file, "r") as f:
num_hidden_layers = json.load(f).get("num_hidden_layers")
if not isinstance(num_hidden_layers, int):
raise RuntimeError(
f"SGLANG_APPLY_CONFIG_BACKUP=auto could not read a numeric "
f"num_hidden_layers from {real_config_file} (got {num_hidden_layers!r})."
)
backup_mode = "small" if num_hidden_layers <= 50 else "large"
logger.warning(
f"SGLANG_APPLY_CONFIG_BACKUP=auto: checkpoint has "
f"num_hidden_layers={num_hidden_layers}, dispatching to {backup_mode!r}."
)
if backup_mode != "none":
backup_file = {
"small": "config_backup_small.json",
"large": "config_backup_large.json",
}.get(backup_mode)
if backup_file is None:
raise ValueError(
f"SGLANG_APPLY_CONFIG_BACKUP={backup_mode!r} is not recognized; "
f"use 'none' (off), 'small', 'large', or 'auto'."
)
config_file = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"configs",
backup_file,
)
logger.warning(
f"SGLANG_APPLY_CONFIG_BACKUP={backup_mode}: using packaged {config_file} "
f"instead of the checkpoint's config.json at {local_path}."
)
else:
config_file = os.path.join(local_path, "config.json")
if not os.path.exists(config_file):
raise RuntimeError(f"Can't find config file at {config_file}.")
with open(config_file, "r") as f:
config_json = json.load(f)
config_json["architectures"] = [architecture]
config_json["model_type"] = "deepseek_v3"
tmp_path = os.path.join(tempfile.gettempdir(), "_tmp_config_folder")
os.makedirs(tmp_path, exist_ok=True)
unique_path = os.path.join(tmp_path, f"{model_type}_{os.getpid()}")
with open(unique_path, "w") as f:
json.dump(config_json, f)
return AutoConfig.from_pretrained(
unique_path, trust_remote_code=trust_remote_code, revision=revision, **kwargs
)
# Temporary hack for Mistral Large
@lru_cache(maxsize=2)
def _load_mistral_large_3_for_causal_LM(
model_path: str,
trust_remote_code: bool = False,
revision: Optional[str] = None,
):
# first get the local path
local_path = download_from_hf(model_path)
# then load the config file in json
parser = mistral_utils.MistralConfigParser()
config_dict, _ = parser.parse(local_path)
with tempfile.NamedTemporaryFile(mode="w+", suffix=".json") as f:
json.dump(config_dict, f)
f.flush()
loaded_config = AutoConfig.from_pretrained(
f.name, trust_remote_code=trust_remote_code, revision=revision
)
text_config = getattr(loaded_config, "text_config", None)
if text_config is not None and isinstance(text_config, dict):
text_config = AutoConfig.for_model(**text_config)
setattr(loaded_config, "text_config", text_config)
vision_config = getattr(loaded_config, "vision_config", None)
if vision_config is not None and isinstance(vision_config, dict):
vision_config = AutoConfig.for_model(**vision_config)
setattr(loaded_config, "vision_config", vision_config)
return loaded_config
def _is_deepseek_ocr_model(config: PretrainedConfig) -> bool:
# TODO: Remove this workaround related when AutoConfig correctly identifies deepseek-ocr.
# Hugging Face's AutoConfig currently misidentifies it as deepseekvl2.
auto_map = getattr(config, "auto_map", None) or {}
return auto_map.get("AutoModel") == "modeling_deepseekocr.DeepseekOCRForCausalLM"
def _is_deepseek_ocr2_model(config: PretrainedConfig) -> bool:
auto_map = getattr(config, "auto_map", None) or {}
return auto_map.get("AutoModel") == "modeling_deepseekocr2.DeepseekOCR2ForCausalLM"
def _override_deepseek_ocr_v_head_dim(config: DeepseekVLV2Config) -> None:
# FIXME: deepseek-ocr's v_head_dim is set to 0 in its config file.
# https://huggingface.co/deepseek-ai/DeepSeek-OCR/blob/main/config.json#L116
if config.text_config.v_head_dim == 0:
V_HEAD_DIM_PATCH = 128
config.text_config.v_head_dim = V_HEAD_DIM_PATCH
# Also fix language_config so get_hf_text_config (which may prefer it
# over text_config) stays consistent.
lc = getattr(config, "language_config", None)
if isinstance(lc, dict):
lc["v_head_dim"] = V_HEAD_DIM_PATCH
elif hasattr(lc, "v_head_dim"):
lc.v_head_dim = V_HEAD_DIM_PATCH
logger.warning(
f"Overriding deepseek-ocr's v_head_dim from 0 to {V_HEAD_DIM_PATCH} to avoid potential issues."
)
def _override_v_head_dim_if_zero(config: PretrainedConfig, patch: int = 128) -> None:
text_config = getattr(config, "text_config", None)
language_config = getattr(config, "language_config", None)
target = text_config or language_config
if target is None:
return
if getattr(target, "v_head_dim", None) == 0:
setattr(target, "v_head_dim", patch)
logger.warning(
f"Overriding v_head_dim from 0 to {patch} to avoid potential issues."
)
def _ensure_clean_up_tokenization_compat() -> None:
"""Re-add ``clean_up_tokenization`` removed in transformers v5.
Remote-code tokenizers (e.g. InternLM2Tokenizer) call
``self.clean_up_tokenization()`` which was a static method on
``PreTrainedTokenizerBase`` in v4 but removed in v5. Patch it back
so existing HuggingFace Hub tokenizer code keeps working.
"""
if hasattr(PreTrainedTokenizerBase, "clean_up_tokenization"):
return
@staticmethod
def clean_up_tokenization(out_string: str) -> str:
out_string = (
out_string.replace(" .", ".")
.replace(" ?", "?")
.replace(" !", "!")
.replace(" ,", ",")
.replace(" ' ", "'")
.replace(" n't", "n't")
.replace(" 'm", "'m")
.replace(" 's", "'s")
.replace(" 've", "'ve")
.replace(" 're", "'re")
)
return out_string
PreTrainedTokenizerBase.clean_up_tokenization = clean_up_tokenization
# Apply immediately so all code paths (get_tokenizer, get_processor,
# and any external callers) benefit without needing an explicit call.
_ensure_clean_up_tokenization_compat()
def _ensure_is_torch_fx_available_compat() -> None:
"""Re-add ``is_torch_fx_available`` removed in transformers v5.
Remote-code models (e.g. MiniCPM-V) import ``is_torch_fx_available``
from ``transformers.utils.import_utils``. The function was removed
in v5. Patch it back so existing HuggingFace Hub model code keeps
working. torch.fx is always available in PyTorch >= 2.0.
"""
import transformers.utils.import_utils as _import_utils
if hasattr(_import_utils, "is_torch_fx_available"):
return
_import_utils.is_torch_fx_available = lambda: True
_ensure_is_torch_fx_available_compat()
def normalize_rope_scaling_compat(config: "PretrainedConfig") -> None:
"""Ensure rope_scaling dicts have ``"type"`` alongside ``"rope_type"``.
Transformers v5 standardises rope_scaling to use ``"rope_type"`` and may
omit the legacy ``"type"`` key. Remote-code models (e.g. Kimi-VL) still
read ``rope_scaling["type"]``, causing a ``KeyError``. This helper adds
``"type"`` from ``"rope_type"`` whenever it is missing, recursively across
the config and all its sub-configs.
"""
def _patch(cfg):
try:
rs = getattr(cfg, "rope_scaling", None)
except AttributeError:
rs = None
if isinstance(rs, dict) and "rope_type" in rs and "type" not in rs:
rs["type"] = rs["rope_type"]
# Recurse into sub-configs
for attr in (
"text_config",
"llm_config",
"language_config",
"vision_config",
"thinker_config",
):
sub = getattr(cfg, attr, None)
if sub is not None:
_patch(sub)
_patch(config)
def _ensure_llama_flash_attention2_compat() -> None:
"""Ensure LlamaFlashAttention2 symbol exists for remote code compatibility."""
try:
from transformers.models.llama import modeling_llama
except (ImportError, ModuleNotFoundError):
return
if not hasattr(modeling_llama, "LlamaFlashAttention2"):
if hasattr(modeling_llama, "LlamaAttention"):
modeling_llama.LlamaFlashAttention2 = modeling_llama.LlamaAttention
def _ensure_gguf_version():
"""Workaround for transformers v5 bug where is_gguf_available() fails
when the gguf package lacks __version__ and metadata lookup also fails,
resulting in packaging.version.InvalidVersion: Invalid version: 'N/A'."""
try:
import gguf
if not hasattr(gguf, "__version__"):
import importlib.metadata
try:
gguf.__version__ = importlib.metadata.version("gguf")
except Exception:
gguf.__version__ = "0.0.0"
except ImportError:
pass
@lru_cache_frozenset(maxsize=32)
def get_config(
model: str,
trust_remote_code: bool,
revision: Optional[str] = None,
model_override_args: Optional[dict] = None,
**kwargs,
):
is_gguf = check_gguf_file(model)
if is_gguf:
_ensure_gguf_version()
kwargs["gguf_file"] = model
model = Path(model).parent
if is_runai_obj_uri(model):
model = ObjectStorageModel.get_path(model)
if is_remote_url(model):
# BaseConnector implements __del__() to clean up the local dir.
# Since config files need to exist all the time, so we DO NOT use
# with statement to avoid closing the client.
client = create_remote_connector(model)
client.pull_files(ignore_pattern=["*.pt", "*.safetensors", "*.bin"])
model = client.get_local_dir()
if (
"mistral-large-3" in str(model).lower()
or "mistral-small-4" in str(model).lower()
or "leanstral" in str(model).lower()
):
config = _load_mistral_large_3_for_causal_LM(
model, trust_remote_code=trust_remote_code, revision=revision
)
elif envs.SGLANG_APPLY_CONFIG_BACKUP.get() != "none":
config = _load_deepseek_temp_model(
model,
model_type="deepseek_ref",
architecture="DeepseekV4ForCausalLM",
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
else:
_ensure_llama_flash_attention2_compat()
try:
config = AutoConfig.from_pretrained(
model, trust_remote_code=trust_remote_code, revision=revision, **kwargs
)
except ValueError as e:
if "deepseek_ref" in str(e):
config = _load_deepseek_temp_model(
model,
model_type="deepseek_ref",
architecture="DeepseekV4ForCausalLM",
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
elif "deepseek_v32" in str(e):
config = _load_deepseek_temp_model(
model,
model_type="deepseek_v32",
architecture="DeepseekV3ForCausalLM",
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
else:
raise e
except KeyError as e:
# Transformers v5 may register a built-in config class that
# conflicts with sglang's custom one (e.g. NemotronHConfig
# doesn't handle '-' in hybrid_override_pattern). Fall back
# to loading the raw config dict and using sglang's class.
# Also handle deepseek_v32 which v5 doesn't recognize.
if "deepseek_v32" in str(e):
config = _load_deepseek_temp_model(
model,
model_type="deepseek_v32",
architecture="DeepseekV3ForCausalLM",
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
else:
config_dict, _ = PretrainedConfig.get_config_dict(
model,
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
model_type = config_dict.get("model_type")
if model_type in _CONFIG_REGISTRY:
config = _CONFIG_REGISTRY[model_type].from_dict(config_dict)
config._name_or_path = model
else:
raise
if (
config.architectures is not None
and config.architectures[0] == "Phi4MMForCausalLM"
):
# Phi4MMForCausalLM uses a hard-coded vision_config. See:
# https://github.com/vllm-project/vllm/blob/6071e989df1531b59ef35568f83f7351afb0b51e/vllm/model_executor/models/phi4mm.py#L71
# We set it here to support cases where num_attention_heads is not divisible by the TP size.
from transformers import SiglipVisionConfig
vision_config = {
"hidden_size": 1152,
"image_size": 448,
"intermediate_size": 4304,
"model_type": "siglip_vision_model",
"num_attention_heads": 16,
"num_hidden_layers": 26,
# Model is originally 27-layer, we only need the first 26 layers for feature extraction.
"patch_size": 14,
}
config.vision_config = SiglipVisionConfig(**vision_config)
if config.architectures in [
["LongcatCausalLM"],
["LongcatFlashForCausalLM"],
["LongcatFlashNgramForCausalLM"],
]:
config.model_type = "longcat_flash"
text_config = get_hf_text_config(config=config)
if isinstance(model, str) and text_config is not None:
items = (
text_config.items()
if hasattr(text_config, "items")
else vars(text_config).items()
)
for key, val in items:
if not hasattr(config, key) and val is not None:
setattr(config, key, val)
if _is_deepseek_ocr2_model(config):
_override_v_head_dim_if_zero(config)
# Temporary hack for load deepseek-ocr2
config.model_type = "deepseek-ocr"
config.update({"architectures": ["DeepseekOCRForCausalLM"]})
config = DeepseekVLV2Config.from_pretrained(model, revision=revision)
_override_v_head_dim_if_zero(config)
config.update({"architectures": ["DeepseekOCRForCausalLM"]})
setattr(config, "_name_or_path", model)
elif config.model_type in _CONFIG_REGISTRY:
model_type = config.model_type
if model_type == "deepseek_vl_v2":
if _is_deepseek_ocr_model(config) or _is_deepseek_ocr2_model(config):
model_type = "deepseek-ocr"
config_class = _CONFIG_REGISTRY[model_type]
config = config_class.from_pretrained(model, revision=revision)
if _is_deepseek_ocr_model(config):
_override_deepseek_ocr_v_head_dim(config)
config.update({"architectures": ["DeepseekOCRForCausalLM"]})
elif _is_deepseek_ocr2_model(config):
_override_v_head_dim_if_zero(config)
config.update({"architectures": ["DeepseekOCRForCausalLM"]})
# NOTE(HandH1998): Qwen2VL requires `_name_or_path` attribute in `config`.
setattr(config, "_name_or_path", model)
if isinstance(model, str) and config.model_type == "internvl_chat":
for key, val in config.llm_config.__dict__.items():
if not hasattr(config, key):
setattr(config, key, val)
if config.model_type == "multi_modality":
config.update({"architectures": ["MultiModalityCausalLM"]})
if config.model_type == "gemma4":
# Gemma4 configs use base attributes for SWA layers and `global_*`
# variants for full-attention layers. SGLang expects the opposite:
# base = full-attention, `swa_*` = sliding-window overrides.
# Remap here so the rest of the stack sees a uniform convention.
text_config = config.text_config
global_head_dim = getattr(text_config, "global_head_dim", None)
global_kv_heads = getattr(text_config, "num_global_key_value_heads", None)
swa_head_dim = text_config.head_dim
swa_kv_heads = text_config.num_key_value_heads
text_config.swa_head_dim = swa_head_dim
text_config.swa_v_head_dim = swa_head_dim
text_config.swa_num_key_value_heads = swa_kv_heads
if global_head_dim is not None:
text_config.head_dim = global_head_dim
if global_kv_heads is not None:
text_config.num_key_value_heads = global_kv_heads
if not hasattr(text_config, "v_head_dim"):
text_config.v_head_dim = text_config.head_dim
if not hasattr(text_config, "swa_v_head_dim"):
text_config.swa_v_head_dim = text_config.swa_head_dim
if config.model_type == "longcat_flash":
config.update({"architectures": ["LongcatFlashForCausalLM"]})
if model_override_args:
config.update(model_override_args)
# Special architecture mapping check for GGUF models
if is_gguf:
if config.model_type not in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES:
raise RuntimeError(f"Can't get gguf config for {config.model_type}.")
model_type = MODEL_FOR_CAUSAL_LM_MAPPING_NAMES[config.model_type]
config.update({"architectures": [model_type]})
return config
@lru_cache_frozenset(maxsize=32)
def get_generation_config(
model: str,
trust_remote_code: bool,
revision: Optional[str] = None,
**kwargs,
):
try:
return GenerationConfig.from_pretrained(
model, trust_remote_code=trust_remote_code, revision=revision, **kwargs
)
except OSError as e:
return None
# Qwen-1M related
def get_sparse_attention_config(
model: str,
sparse_attention_config_filename: str = "sparse_attention_config.json",
) -> Dict[str, Any]:
is_local = os.path.isdir(model)
if not is_local:
# Download the config files.
model = download_from_hf(model, allow_patterns=["*.json"])
config_file = os.path.join(model, sparse_attention_config_filename)
if not os.path.exists(config_file):
return {}
# Load the sparse attention config.
with open(config_file) as f:
config = json.load(f)
return config
# Models don't use the same configuration key for determining the maximum
# context length. Store them here so we can sanely check them.
# NOTE: The ordering here is important. Some models have two of these and we
# have a preference for which value gets used.
CONTEXT_LENGTH_KEYS = [
"max_sequence_length",
"seq_length",
"max_seq_len",
"model_max_length",
"max_position_embeddings",
]
def get_context_length(config):
"""Get the context length of a model from a huggingface model configs."""
text_config = config
rope_scaling = getattr(text_config, "rope_scaling", None)
if rope_scaling:
rope_scaling_factor = rope_scaling.get("factor", 1)
if "original_max_position_embeddings" in rope_scaling:
rope_scaling_factor = 1
if rope_scaling.get("rope_type", None) == "llama3":
rope_scaling_factor = 1
else:
rope_scaling_factor = 1
for key in CONTEXT_LENGTH_KEYS:
val = getattr(text_config, key, None)
if val is not None:
return int(rope_scaling_factor * val)
return 2048
# A fast LLaMA tokenizer with the pre-processed `tokenizer.json` file.
_FAST_LLAMA_TOKENIZER = "hf-internal-testing/llama-tokenizer"
# Filter warnings like: https://github.com/sgl-project/sglang/issues/8082
class TokenizerWarningsFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
return "Calling super().encode with" not in record.getMessage()
_is_base_mistral_patched = False
# transformers version where _patch_mistral_regex calls model_info() on every tokenizer load
_TRANSFORMERS_PATCHED_VERSION = "5.3.0"
def _patch_is_base_mistral_in_ci():
"""Patch transformers' _patch_mistral_regex to avoid HF API calls in CI.
transformers defines is_base_mistral as a local function inside
_patch_mistral_regex, so it cannot be patched via module attribute.
Instead we replace the entire _patch_mistral_regex classmethod with a
version that simply returns the tokenizer unchanged.
In CI this prevents exhausting the 3000 req/5min HF API rate limit.
"""
global _is_base_mistral_patched
if _is_base_mistral_patched:
return
from sglang.srt.environ import envs
if not envs.SGLANG_IS_IN_CI.get():
return
import transformers
if transformers.__version__ != _TRANSFORMERS_PATCHED_VERSION:
logger.warning(
"transformers version changed to %s (expected %s), "
"_patch_mistral_regex patch skipped — may need update if 429 errors recur",
transformers.__version__,
_TRANSFORMERS_PATCHED_VERSION,
)
_is_base_mistral_patched = True # don't warn repeatedly
return
from transformers import PreTrainedTokenizerFast
if hasattr(PreTrainedTokenizerFast, "_patch_mistral_regex"):
@classmethod
def _noop_patch_mistral_regex(cls, tokenizer, *args, **kwargs):
return tokenizer
PreTrainedTokenizerFast._patch_mistral_regex = _noop_patch_mistral_regex
logger.info("CI: patched _patch_mistral_regex to skip HF API calls")
_is_base_mistral_patched = True
def get_tokenizer(
tokenizer_name: str,
*args,
tokenizer_mode: str = "auto",
trust_remote_code: bool = False,
tokenizer_revision: Optional[str] = None,
**kwargs,
) -> Union[PreTrainedTokenizer, PreTrainedTokenizerFast]:
"""Gets a tokenizer for the given model name via Huggingface."""
if tokenizer_name.endswith(".json"):
from sglang.srt.tokenizer.tiktoken_tokenizer import TiktokenTokenizer
return TiktokenTokenizer(tokenizer_name)
if tokenizer_mode == "slow":
if kwargs.get("use_fast", False):
raise ValueError("Cannot use the fast tokenizer in slow tokenizer mode.")
kwargs["use_fast"] = False
elif tokenizer_mode == "auto":
# In Transformers v5, the default for use_fast changed from True to False.
# Explicitly set use_fast=True for "auto" mode to maintain previous behavior
# and avoid issues with models that have incorrect tokenizer_class values.
if "use_fast" not in kwargs:
kwargs["use_fast"] = True
# TODO(Xinyuan): Remove this once we have a proper tokenizer for Devstral
if tokenizer_name == "mistralai/Devstral-Small-2505":
tokenizer_name = "mistralai/Mistral-Small-3.1-24B-Instruct-2503"
is_gguf = check_gguf_file(tokenizer_name)
if is_gguf:
_ensure_gguf_version()
kwargs["gguf_file"] = tokenizer_name
tokenizer_name = Path(tokenizer_name).parent
if is_runai_obj_uri(tokenizer_name):
tokenizer_name = ObjectStorageModel.get_path(tokenizer_name)
if is_remote_url(tokenizer_name):
# BaseConnector implements __del__() to clean up the local dir.
# Since config files need to exist all the time, so we DO NOT use
# with statement to avoid closing the client.
client = create_remote_connector(tokenizer_name)
client.pull_files(ignore_pattern=["*.pt", "*.safetensors", "*.bin"])
tokenizer_name = client.get_local_dir()
_patch_is_base_mistral_in_ci()
try:
tokenizer = AutoTokenizer.from_pretrained(
tokenizer_name,
*args,
trust_remote_code=trust_remote_code,
tokenizer_revision=tokenizer_revision,
clean_up_tokenization_spaces=False,
**kwargs,
)
# Filter tokenizer warnings
logging.getLogger(tokenizer.__class__.__module__).addFilter(
TokenizerWarningsFilter()
)
except TypeError as e:
# The LLaMA tokenizer causes a protobuf error in some environments.
err_msg = (
"Failed to load the tokenizer. If you are using a LLaMA V1 model "
f"consider using '{_FAST_LLAMA_TOKENIZER}' instead of the "
"original tokenizer."
)
raise RuntimeError(err_msg) from e
except ValueError as e:
# If the error pertains to the tokenizer class not existing or not
# currently being imported, suggest using the --trust-remote-code flag.
if not trust_remote_code and (
"does not exist or is not currently imported." in str(e)
or "requires you to execute the tokenizer file" in str(e)
):
err_msg = (
"Failed to load the tokenizer. If the tokenizer is a custom "
"tokenizer not yet available in the HuggingFace transformers "
"library, consider setting `trust_remote_code=True` in LLM "
"or using the `--trust-remote-code` flag in the CLI."
)
raise RuntimeError(err_msg) from e
else:
raise e
# Transformers v5 may silently fall back to a generic TokenizersBackend
# when trust_remote_code=False and the model requires a custom tokenizer.
# Detect this and auto-retry with trust_remote_code=True.
if not trust_remote_code and type(tokenizer).__name__ == "TokenizersBackend":
tokenizer = AutoTokenizer.from_pretrained(
tokenizer_name,
*args,
trust_remote_code=True,
tokenizer_revision=tokenizer_revision,
clean_up_tokenization_spaces=False,
**kwargs,
)
_fix_v5_tokenizer_components(tokenizer, tokenizer_name, tokenizer_revision)
_fix_v5_add_bos_eos_token(tokenizer, tokenizer_name, tokenizer_revision)
if not isinstance(tokenizer, PreTrainedTokenizerFast):
warnings.warn(
"Using a slow tokenizer. This might cause a significant "
"slowdown. Consider using a fast tokenizer instead."
)
_fix_special_tokens_pattern(tokenizer)
attach_additional_stop_token_ids(tokenizer)
return tokenizer
def _resolve_local_or_cached_file(model_name_or_path, filename, revision=None):
"""Resolve a file from a local directory or HF hub cache (no network)."""
local_path = Path(model_name_or_path) / filename
if local_path.is_file():
return str(local_path)
from huggingface_hub import hf_hub_download
return hf_hub_download(
model_name_or_path, filename, revision=revision, local_files_only=True
)
def _fix_v5_tokenizer_components(tokenizer, model_name_or_path, revision=None):
"""Fix pre_tokenizer/decoder when a v5 tokenizer class overwrites them.
In transformers v5, some tokenizer classes (e.g. LlamaTokenizer) have a
custom __init__ that rebuilds the pre_tokenizer and decoder from scratch
with class-specific components, discarding the originals from tokenizer.json.
This breaks models that specify LlamaTokenizerFast but actually use a
different tokenizer architecture (e.g. DeepSeek-V3.2 uses ByteLevel).
Detects the mismatch by comparing against the raw tokenizer.json and
restores the original components when they differ.
"""
backend = getattr(tokenizer, "_tokenizer", None)
if backend is None:
return
try:
from tokenizers import Tokenizer as RawTokenizer
tok_file = _resolve_local_or_cached_file(
model_name_or_path, "tokenizer.json", revision
)
raw = RawTokenizer.from_file(tok_file)
except Exception as e:
logger.debug(
"_fix_v5_tokenizer_components: could not load tokenizer.json for %s: %s",
model_name_or_path,
e,
)
return
raw_pre = type(raw.pre_tokenizer).__name__ if raw.pre_tokenizer else None
loaded_pre = type(backend.pre_tokenizer).__name__ if backend.pre_tokenizer else None
if raw_pre and loaded_pre and raw_pre != loaded_pre:
logger.info(
"Fixing v5 tokenizer component mismatch for %s: "
"pre_tokenizer %s -> %s, decoder %s -> %s",
model_name_or_path,
loaded_pre,
raw_pre,
type(backend.decoder).__name__ if backend.decoder else None,
type(raw.decoder).__name__ if raw.decoder else None,
)
backend.pre_tokenizer = raw.pre_tokenizer
backend.decoder = raw.decoder
def _fix_v5_add_bos_eos_token(tokenizer, model_name_or_path, revision=None):
"""Restore add_bos_token/add_eos_token stripped by transformers v5.
In transformers v5, _from_pretrained() strips add_bos_token and
add_eos_token from init kwargs when a tokenizer.json file is present,
assuming the tokenizer.json post-processor handles BOS/EOS addition.
However, many models (e.g. DeepSeek-V3) have a tokenizer.json whose
post-processor does NOT add BOS/EOS, and rely on the add_bos_token flag
from tokenizer_config.json instead. This causes silent accuracy regressions.
This function reads the tokenizer_config.json and restores the values,
but only for tokenizer classes that actually supported these flags in v4.
Classes like Qwen2Tokenizer did not support add_bos_token/add_eos_token
in v4, so restoring them would change behavior.
"""
# In transformers v4, only certain tokenizer classes supported
# add_bos_token / add_eos_token as init parameters. Restoring these
# flags for classes that never supported them (e.g. Qwen2Tokenizer)
# would incorrectly change tokenization behavior.
_V4_CLASSES_WITH_BOS_EOS_FLAGS = frozenset(
{
"LlamaTokenizer",
"LlamaTokenizerFast",
"CodeLlamaTokenizer",
"CodeLlamaTokenizerFast",
"GemmaTokenizer",
"GemmaTokenizerFast",
"CohereTokenizerFast",
}
)
try:
config_file = _resolve_local_or_cached_file(
model_name_or_path, "tokenizer_config.json", revision
)
with open(config_file) as f:
config = json.load(f)
except Exception as e:
logger.debug(
"_fix_v5_add_bos_eos_token: could not read tokenizer_config.json "
"for %s: %s",
model_name_or_path,
e,
)
return
tokenizer_class = config.get("tokenizer_class", "")
if tokenizer_class not in _V4_CLASSES_WITH_BOS_EOS_FLAGS:
logger.debug(
"_fix_v5_add_bos_eos_token: skipping %s (tokenizer_class=%s "
"did not support add_bos/eos_token in v4)",
model_name_or_path,
tokenizer_class,
)
return
# In v4, Llama/Gemma tokenizers defaulted add_bos_token=True.
# When the config omits the key or has null, use the v4 default so that
# update_post_processor() doesn't drop BOS/EOS that was there before.
_V4_DEFAULTS = {"add_bos_token": True, "add_eos_token": False}
changed = False
for attr in ("add_bos_token", "add_eos_token"):
config_val = config.get(attr)
if config_val is None:
# Key missing or null → use v4 default for this tokenizer class
config_val = _V4_DEFAULTS.get(attr, False)
# Fast tokenizers in v4 used tokenizer.json post-processor for EOS —
# the add_eos_token Python attribute was set but the post-processor
# came from tokenizer.json, not from the attribute. In v5, the flag is
# stripped and both sglang and HF reference end up with add_eos_token=False.
# Restoring add_eos_token for fast tokenizers makes sglang diverge from
# the HF reference (which doesn't restore it), breaking embedding models
# like intfloat/e5-mistral-7b-instruct (cosine similarity drops to ~0.33).
if attr == "add_eos_token" and isinstance(tokenizer, PreTrainedTokenizerFast):
config_val = _V4_DEFAULTS["add_eos_token"] # False
current_val = getattr(tokenizer, attr, None)
if current_val != config_val:
logger.info(
"Restoring %s=%s for %s (was %s after v5 loading)",
attr,
config_val,
model_name_or_path,
current_val,
)
setattr(tokenizer, f"_{attr}", config_val)
changed = True
# Rebuild the post-processor so it respects the restored flags
if changed and hasattr(tokenizer, "update_post_processor"):
tokenizer.update_post_processor()
def _fix_special_tokens_pattern(tokenizer):
"""Fix https://github.com/huggingface/transformers/pull/42563 which defaults
special_tokens_pattern to "cls_sep", inserting None into token IDs when
cls_token/sep_token are undefined (e.g. Kimi-VL's TikTokenTokenizer).
"""
pattern = getattr(tokenizer, "special_tokens_pattern", None)
if pattern == "cls_sep" and (
tokenizer.cls_token_id is None or tokenizer.sep_token_id is None
):
tokenizer.special_tokens_pattern = "none"
def _fix_added_tokens_encoding(tokenizer):
"""Ensure special tokens encode as single tokens in transformers v5.
Some model tokenizers (e.g. MiniCPM-V-4) define special tokens like <image>,
<slice> as attributes on the tokenizer class with corresponding IDs in the
vocabulary (via tokenizer.json's added_tokens). In transformers v5, these
tokens may not appear in get_added_vocab() and encode() splits them into
subwords, breaking multimodal pipelines that rely on finding them in input_ids.
This function discovers such tokens by scanning tokenizer attributes, checks
if they encode correctly, and re-registers any that don't.
"""
# Discover special token strings from tokenizer attributes.
# Model tokenizers (e.g. MiniCPMVTokenizerFast) store them as attributes
# like im_start="<image>", slice_start="<slice>", etc.
candidates = {}
for attr in dir(tokenizer):
if attr.startswith("_"):
continue
try:
val = getattr(tokenizer, attr)
except Exception:
continue
if (
not isinstance(val, str)
or not val.startswith("<")
or not val.endswith(">")
or len(val) > 20
):
continue
token_id = tokenizer.convert_tokens_to_ids(val)
if token_id is not None and token_id != tokenizer.unk_token_id:
candidates[val] = token_id
if not candidates:
return
# Check which tokens fail to encode as single tokens.
broken = []
for token_str, expected_id in candidates.items():
try:
ids = tokenizer.encode(token_str, add_special_tokens=False)
if len(ids) != 1 or ids[0] != expected_id:
broken.append(token_str)
except Exception:
broken.append(token_str)
if not broken:
return
from transformers import AddedToken
tokens_to_add = [AddedToken(tok, special=True, normalized=False) for tok in broken]
tokenizer.add_tokens(tokens_to_add, special_tokens=True)
logger.info(
"Re-registered %d special tokens for correct v5 encoding: %s",
len(broken),
broken[:10],
)
# Some models doesn't have an available processor, e.g.: InternVL
def get_tokenizer_from_processor(processor):
if isinstance(processor, PreTrainedTokenizerBase):
return processor
return processor.tokenizer
def _build_processor_manually(
model_path, config, trust_remote_code, revision, **kwargs
):
"""Build processor when AutoProcessor fails to resolve feature_extractor_type.
In transformers v5, AutoProcessor.from_pretrained calls
AutoFeatureExtractor.from_pretrained which fails if
preprocessor_config.json lacks 'feature_extractor_type'. This loads the
processor class from the hub and constructs it with individually-loaded
components.
"""
import transformers
from transformers import AutoImageProcessor, AutoTokenizer
from transformers.dynamic_module_utils import get_class_from_dynamic_module
# Resolve processor class from auto_map — check both the model config
# and the preprocessor_config.json (some models like MiniCPM-o only
# declare AutoProcessor in the latter).
auto_map = getattr(config, "auto_map", None) or {}
proc_ref = auto_map.get("AutoProcessor")
if not proc_ref:
try:
pp_file = _resolve_local_or_cached_file(
model_path, "preprocessor_config.json", revision
)
with open(pp_file) as f:
pp_auto_map = json.load(f).get("auto_map", {})
proc_ref = pp_auto_map.get("AutoProcessor")
except Exception as e:
logger.debug(
"_build_processor_manually: could not read preprocessor_config.json "
"for %s: %s",
model_path,
e,
)
if not proc_ref:
raise ValueError(f"Cannot determine processor class for {model_path}")
proc_cls = get_class_from_dynamic_module(
proc_ref, model_path, code_revision=revision
)
# Load sub-components individually (these succeed)
tokenizer = AutoTokenizer.from_pretrained(
model_path, trust_remote_code=trust_remote_code, revision=revision
)
init_kwargs = {"tokenizer": tokenizer}
if "image_processor" in getattr(proc_cls, "attributes", []):
try:
init_kwargs["image_processor"] = AutoImageProcessor.from_pretrained(
model_path, trust_remote_code=trust_remote_code, revision=revision
)
except Exception as e:
logger.warning("Failed to load image_processor for %s: %s", model_path, e)
# Instantiate feature extractor from its declared class
fe_class_name = getattr(proc_cls, "feature_extractor_class", None)
if fe_class_name:
fe_class = getattr(transformers, fe_class_name, None)
if fe_class is not None:
init_kwargs["feature_extractor"] = fe_class()
return proc_cls(**init_kwargs)
def get_processor(
tokenizer_name: str,
*args,
tokenizer_mode: str = "auto",
trust_remote_code: bool = False,
tokenizer_revision: Optional[str] = None,
use_fast: Optional[bool] = True,
**kwargs,
):
# pop 'revision' from kwargs if present.
revision = kwargs.pop("revision", tokenizer_revision)
if (
"mistral-large-3" in str(tokenizer_name).lower()
or "mistral-small-4" in str(tokenizer_name).lower()
or "leanstral" in str(tokenizer_name).lower()
):
config = _load_mistral_large_3_for_causal_LM(
tokenizer_name,
trust_remote_code=trust_remote_code,
revision=revision,
)
else:
_ensure_llama_flash_attention2_compat()
config = AutoConfig.from_pretrained(
tokenizer_name,
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
if _is_deepseek_ocr_model(config):
# Temporary hack for load deepseek-ocr
config.model_type = "deepseek-ocr"
config.update({"architectures": ["DeepseekOCRForCausalLM"]})
elif _is_deepseek_ocr2_model(config):
# Temporary hack for load deepseek-ocr2
config.model_type = "deepseek-ocr"
config.update({"architectures": ["DeepseekOCRForCausalLM"]})
_override_v_head_dim_if_zero(config)
# fix: for Qwen2-VL and Sarashina2Vision models, inject default 'size' if not provided.
if config.model_type in {"qwen2_vl", "sarashina2_vision"}:
if "size" not in kwargs:
kwargs["size"] = {"shortest_edge": 3136, "longest_edge": 1003520}
if config.model_type not in {"llava", "clip"}:
kwargs["use_fast"] = use_fast
try:
if "InternVL3_5" in tokenizer_name:
processor = AutoTokenizer.from_pretrained(
tokenizer_name,
*args,
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
else:
if config.model_type in _CUSTOMIZED_MM_PROCESSOR:
processor = _CUSTOMIZED_MM_PROCESSOR[config.model_type].from_pretrained(
tokenizer_name,
*args,
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
else:
processor = AutoProcessor.from_pretrained(
tokenizer_name,
*args,
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
except ValueError as e:
error_message = str(e)
if "does not have a slow version" in error_message:
logger.info(
f"Processor {tokenizer_name} does not have a slow version. Automatically use fast version"
)
kwargs["use_fast"] = True
processor = AutoProcessor.from_pretrained(
tokenizer_name,
*args,
trust_remote_code=trust_remote_code,
revision=revision,
**kwargs,
)
elif "Unrecognized feature extractor" in error_message:
logger.info(
"AutoProcessor failed on feature extractor for %s, "
"constructing processor manually",
tokenizer_name,
)
processor = _build_processor_manually(
tokenizer_name,
config,
trust_remote_code,
revision,
**kwargs,
)
else:
raise e
# If processor is a bare tokenizer (e.g. Mistral-Small-4 has no processor_config.json)
# and the model is a vision model (pixtral), wrap it in a proper PixtralProcessor
# so that image data is actually processed through the image processor.
if (
isinstance(processor, PreTrainedTokenizerBase)
and getattr(config, "model_type", None) == "pixtral"
):
from transformers.models.pixtral.image_processing_pixtral import (
PixtralImageProcessor,
)
from transformers.models.pixtral.processing_pixtral import (
PixtralProcessor as HFPixtralProcessor,
)
vision_config = config.vision_config
patch_size = vision_config.patch_size
image_size = vision_config.image_size
spatial_merge_size = getattr(vision_config, "spatial_merge_size", 1)
effective_patch = patch_size * spatial_merge_size
image_processor = PixtralImageProcessor(
do_resize=True,
size={"longest_edge": image_size},
patch_size={"height": effective_patch, "width": effective_patch},
)
processor = HFPixtralProcessor(
image_processor=image_processor,
tokenizer=processor,
patch_size=patch_size,
spatial_merge_size=spatial_merge_size,
)
tokenizer = get_tokenizer_from_processor(processor)
if tokenizer.chat_template is None:
local_path = download_from_hf(
tokenizer_name, allow_patterns=["*.json", "*.jinja", "*.model"]
)
jinja_path = Path(local_path) / "chat_template.jinja"
if jinja_path.is_file():
tokenizer.chat_template = jinja_path.read_text()
logger.info("Loaded chat_template from %s", jinja_path)
_fix_special_tokens_pattern(tokenizer)
_fix_added_tokens_encoding(tokenizer)
attach_additional_stop_token_ids(tokenizer)
return processor
def attach_additional_stop_token_ids(tokenizer):
# Special handling for stop token <|eom_id|> generated by llama 3 tool use.
if "<|eom_id|>" in tokenizer.get_added_vocab():
tokenizer.additional_stop_token_ids = set(
[tokenizer.get_added_vocab()["<|eom_id|>"]]
)
else:
tokenizer.additional_stop_token_ids = None
def check_gguf_file(model: Union[str, os.PathLike]) -> bool:
"""Check if the file is a GGUF model."""
model = Path(model)
if not model.is_file():
return False
elif model.suffix == ".gguf":
return True
with open(model, "rb") as f:
header = f.read(4)
return header == b"GGUF"
Reference in New Issue View Git Blame Copy Permalink