vllm.transformers_utils.repo_utils ¶

def _try_download_from_hf_hub(
    model: str | Path, file_name: str, revision: str | None
) -> Path | None:
    """Try to download a file from HuggingFace Hub.

    Returns the local path on success, None on failure.
    Skips download if model is a local directory.
    """
    if Path(model).is_dir():
        return None
    try:
        return Path(
            hf_api().hf_hub_download(
                model,
                file_name,
                revision=revision,
            )
        )
    except huggingface_hub.errors.OfflineModeIsEnabled:
        return None
    except (
        RepositoryNotFoundError,
        RevisionNotFoundError,
        EntryNotFoundError,
        LocalEntryNotFoundError,
    ) as e:
        logger.debug("File or repository not found in hf_hub_download: %s", e)
        return None
    except HfHubHTTPError as e:
        logger.warning(
            "Cannot connect to Hugging Face Hub. Skipping file download for '%s':",
            file_name,
            exc_info=e,
        )
        return None

def get_hf_file_bytes(
    file_name: str, model: str | Path, revision: str | None = "main"
) -> bytes | None:
    """Get file contents from HuggingFace repository as bytes."""
    file_path = try_get_local_file(model=model, file_name=file_name, revision=revision)

    if file_path is None:
        file_path = _try_download_from_hf_hub(model, file_name, revision)

    if isinstance(file_path, Path) and file_path.is_file():
        with open(file_path, "rb") as file:
            return file.read()

    return None

def get_hf_file_to_dict(
    file_name: str, model: str | Path, revision: str | None = "main"
):
    """
    Downloads a file from the Hugging Face Hub and returns
    its contents as a dictionary.

    Parameters:
    - file_name (str): The name of the file to download.
    - model (str): The name of the model on the Hugging Face Hub.
    - revision (str): The specific version of the model.

    Returns:
    - config_dict (dict): A dictionary containing
    the contents of the downloaded file.
    """

    file_path = try_get_local_file(model=model, file_name=file_name, revision=revision)

    if file_path is None:
        file_path = _try_download_from_hf_hub(model, file_name, revision)

    if isinstance(file_path, Path) and file_path.is_file():
        with open(file_path) as file:
            return json.load(file)

    return None

def hf_api() -> HfApi:
    """Return a shared HfApi instance tagged with vLLM's library info."""
    global _hf_api
    if _hf_api is None:
        _hf_api = HfApi(
            library_name="vllm",
            library_version=VLLM_VERSION,
        )
    return _hf_api

def hf_fs() -> "huggingface_hub.HfFileSystem":
    """Return a fresh HfFileSystem tagged with vLLM's library info."""
    return huggingface_hub.HfFileSystem(
        library_name="vllm",
        library_version=VLLM_VERSION,
    )

def resolve_revision(
    repo_id: str,
    revision: str | None = None,
    token: str | bool | None = None,
) -> str | None:
    """Best-effort attempt to resolve a Hub revision to a commit hash.

    Resolving the commit hash once prevents repeated HTTP calls downstream and
    avoids races if the branch moves while the model is loading.

    `HfApi.resolve_revision` returns a `ResolvedRevision`: a `str` equal to the
    requested revision that also carries the commit hash, so downstream error
    messages stay readable and offline loads reuse the cached `refs/` entry.

    Returns:
        The resolved revision, or `revision` unchanged if it cannot be resolved
        (local path, ModelScope, or any Hub error).
    """
    if Path(repo_id).exists() or envs.VLLM_USE_MODELSCOPE:
        return revision

    try:
        return hf_api().resolve_revision(
            repo_id,
            revision=revision,
            local_files_only=huggingface_hub.constants.HF_HUB_OFFLINE,
            token=token,
        )
    except Exception:
        logger.debug(
            "Failed to resolve revision for %s; falling back to %s.",
            repo_id,
            revision,
            exc_info=True,
        )
        return revision

def try_get_local_file(
    model: str | Path, file_name: str, revision: str | None = "main"
) -> Path | Any | None:
    """
    Try to get a local file from the HuggingFace repository.

    The possible return values are:

    - A `Path` object if the local file is found
    - The `huggingface_hub._CACHED_NO_EXIST` sentinel if the file is known to not exist
    - `None` if the file is not found and we cannot determine if it exists or not

    Callers of this method should handle the `_CACHED_NO_EXIST` sentinel appropriately.
    Checking if the return value `is not None` is not sufficient because it does not
    distinguish between the file not existing and the file not being found.
    """
    file_path = Path(model) / file_name
    if file_path.is_file():
        return file_path
    else:
        try:
            cached_filepath = try_to_load_from_cache(
                repo_id=model, filename=file_name, revision=revision
            )
            if isinstance(cached_filepath, str):
                return Path(cached_filepath)
            return cached_filepath
        except ValueError:
            ...
    return None

def with_retry(
    func: Callable[[], _R],
    log_msg: str,
    max_retries: int = 2,
    retry_delay: int = 2,
    fatal_errors: tuple[type[Exception], ...] = (),
) -> _R:
    """Call `func`, retrying transient failures.

    Args:
        func: The call to make.
        log_msg: Prefix for the message logged when a call fails.
        max_retries: How many times to call `func` before giving up.
        retry_delay: Seconds to wait after the first failure, doubled each time.
        fatal_errors: Exceptions that are a definitive answer rather than a
            transient failure. These are raised immediately, without logging.
    """
    for attempt in range(max_retries):
        try:
            return func()
        except fatal_errors:
            raise
        except Exception as e:
            if attempt == max_retries - 1:
                logger.error("%s: %s", log_msg, e)
                raise
            logger.error(
                "%s: %s, retrying %d of %d", log_msg, e, attempt + 1, max_retries
            )
            time.sleep(retry_delay)
            retry_delay *= 2

    raise AssertionError("Should not be reached")