vllm.v1.structured_output.utils

Classes:

• OutlinesDiskCache –

  SQLite-backed cache for outlines_core.Index objects.

• OutlinesVocabulary –

  Wrapper class for outlines_core.Vocabulary,

Functions:

• compile_regex_with_timeout –

  Run a regex compilation callable with a timeout.

• get_outlines_cache –

  Get the Cache instance to be used for index caching.

• get_outlines_cache_path –

  Get the context object that contains previously-computed return values.

• get_outlines_vocabulary –

  Get the Vocabulary object for a given tokenizer.

• grammar_is_likely_lark –

  Check if grammar appears to use Lark syntax.

• strip_speculative_padding –

  Drop speculative-decoding padding from a token block.

OutlinesDiskCache

SQLite-backed cache for outlines_core.Index objects.

Uses outlines_core's native binary serialization (via Rust serde) instead of pickle, eliminating arbitrary code execution risk on deserialization.

Source code in vllm/v1/structured_output/utils.py

class OutlinesDiskCache:
    """SQLite-backed cache for outlines_core.Index objects.

    Uses outlines_core's native binary serialization (via Rust serde)
    instead of pickle, eliminating arbitrary code execution risk on
    deserialization.
    """

    _TYPE_INDEX = "I"
    _TYPE_STRING = "S"

    def __init__(self, path: str):
        os.makedirs(path, exist_ok=True)
        db_path = os.path.join(path, "outlines_cache.db")
        self._db = sqlite3.connect(db_path, check_same_thread=False)
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute(
            "CREATE TABLE IF NOT EXISTS cache "
            "(key TEXT PRIMARY KEY, type_tag TEXT NOT NULL, value BLOB NOT NULL)"
        )
        self._db.commit()

    def __contains__(self, key: str) -> bool:
        row = self._db.execute("SELECT 1 FROM cache WHERE key=?", (key,)).fetchone()
        return row is not None

    def __getitem__(self, key: str):
        row = self._db.execute(
            "SELECT type_tag, value FROM cache WHERE key=?", (key,)
        ).fetchone()
        if row is None:
            raise KeyError(key)
        type_tag, data = row
        if type_tag == self._TYPE_STRING:
            return data.decode("utf-8")
        return oc.Index.from_binary(data)

    def __setitem__(self, key: str, value):
        if isinstance(value, str):
            type_tag = self._TYPE_STRING
            data = value.encode("utf-8")
        else:
            type_tag = self._TYPE_INDEX
            data = value.__reduce__()[1][0]
        self._db.execute(
            "INSERT OR REPLACE INTO cache (key, type_tag, value) VALUES (?, ?, ?)",
            (key, type_tag, data),
        )
        self._db.commit()

    def get(self, key: str, default=None):
        try:
            return self[key]
        except KeyError:
            return default

    def set(self, key: str, value):
        self[key] = value

    def clear(self):
        self._db.execute("DELETE FROM cache")
        self._db.commit()

OutlinesVocabulary

Wrapper class for outlines_core.Vocabulary, which allows us to store a hash with the vocabulary

Source code in vllm/v1/structured_output/utils.py

class OutlinesVocabulary:
    """Wrapper class for `outlines_core.Vocabulary`,
    which allows us to store a hash with the vocabulary
    """

    def __init__(self, vocabulary: oc.Vocabulary) -> None:
        # Actual vocabulary object
        self.inner = vocabulary
        # Have to do abs(hash()) because python hashes can
        # be negative, and we are using hash as a cache key.
        hex_str = hashlib.sha256(vocabulary.__repr__().encode("utf-8")).hexdigest()
        hash_int = int(hex_str, 16)
        self._hash = hash_int

_reduced_vocabulary(tokenizer)

Create a map from vocabulary tokens to lists of equivalent token ids.

Returns:

• dict[bytes, list[int]] –

  A Dict of token string -> equivalent token ids

Source code in vllm/v1/structured_output/utils.py

def _reduced_vocabulary(tokenizer: TokenizerLike) -> dict[bytes, list[int]]:
    """Create a map from vocabulary tokens to lists of equivalent token ids.

    Returns:
        A Dict of token string -> equivalent token ids

    """
    eos_token_id = tokenizer.eos_token_id

    unicode_to_bytes = {
        v: k for k, v in convert_slow_tokenizer.bytes_to_unicode().items()
    }

    def convert_token_to_string(token: str) -> str:
        string = tokenizer.convert_tokens_to_string([token])

        # A hack to handle missing spaces to HF's Llama tokenizers
        if (
            type(token) is str
            and token.startswith(file_utils.SPIECE_UNDERLINE)
            or token == "<0x20>"
        ):
            return " " + string

        return string

    vocabulary: dict[bytes, list[int]] = {}
    empty_token_ids: list[int] = []
    for token, token_idx in tokenizer.get_vocab().items():
        if token in tokenizer.all_special_tokens:
            continue

        token_str = convert_token_to_string(token)
        if token_str:
            if isinstance(token, (bytes, bytearray)):
                # For BPE tokenizers where tokens are stored as bytes.

                # safe to ignore since token_str is of type (bytearray, bytes)
                # by this point.
                token_bytes = bytes(token_str)  # type: ignore[arg-type]

            elif (token_str == "\ufffd" and token != "\ufffd") or (
                "\ufffd" in token_str and not re_replacement_seq.match(token_str)
            ):
                # Handle tokens with invalid UTF-8 sequences.
                if re_llama_byte_token.match(token):
                    # Llama-like tokenizers use <0xXX> for incomplete sequences.
                    token_bytes = bytes([int(token[3:5], 16)])
                else:
                    # GPT2 tokenizers: map each byte back using unicode_to_bytes
                    byte_vals = [unicode_to_bytes.get(c) for c in token]
                    if None in byte_vals:
                        raise RuntimeError(
                            f"Cannot convert token `{token}`"
                            f" ({token_idx}) to bytes: {token_str}"
                        )
                    # safe to ignore, since if None in byte_vals,
                    # an error is thrown.
                    token_bytes = bytes(byte_vals)  # type: ignore[arg-type]
            else:
                token_bytes = token_str.encode("utf-8")

            if token_idx != eos_token_id:
                vocabulary.setdefault(token_bytes, []).append(token_idx)
        else:
            empty_token_ids.append(token_idx)

    return vocabulary

compile_regex_with_timeout(fn, pattern)

Run a regex compilation callable with a timeout.

Prevents ReDoS attacks where adversarial regex patterns (e.g. nested quantifiers like (a+)+b) cause exponential DFA state-space explosion, hanging the inference worker indefinitely.

Parameters:

• fn

  (Callable[[str], _T]) –

  Single-argument callable that takes the pattern and performs the regex compilation.

• pattern

  (str) –

  The regex pattern string, passed to fn and included in timeout error messages.

Raises:

• ValueError –

  If compilation exceeds the configured timeout.

Source code in vllm/v1/structured_output/utils.py

def compile_regex_with_timeout(fn: Callable[[str], _T], pattern: str) -> _T:
    """Run a regex compilation callable with a timeout.

    Prevents ReDoS attacks where adversarial regex patterns (e.g. nested
    quantifiers like `(a+)+b`) cause exponential DFA state-space explosion,
    hanging the inference worker indefinitely.

    Args:
        fn: Single-argument callable that takes the pattern and performs
            the regex compilation.
        pattern: The regex pattern string, passed to *fn* and included in
            timeout error messages.

    Raises:
        ValueError: If compilation exceeds the configured timeout.

    """
    timeout = envs.VLLM_REGEX_COMPILATION_TIMEOUT_S
    if timeout <= 0:
        return fn(pattern)

    executor = ThreadPoolExecutor(max_workers=1)
    future = executor.submit(fn, pattern)
    try:
        result = future.result(timeout=timeout)
    except TimeoutError:
        future.cancel()
        executor.shutdown(wait=False, cancel_futures=True)
        raise ValueError(
            f"Regex compilation timed out after {timeout}s. "
            "The pattern may be too complex or contain constructs that "
            "cause exponential state-space explosion (e.g. nested "
            f"quantifiers). Pattern: {pattern[:200]}"
        ) from None
    else:
        executor.shutdown(wait=False)
        return result

get_outlines_cache()

Get the Cache instance to be used for index caching.

Source code in vllm/v1/structured_output/utils.py

def get_outlines_cache():
    """Get the Cache instance to be used for index caching."""
    cache_dir = get_outlines_cache_path()
    if envs.VLLM_V1_USE_OUTLINES_CACHE:
        logger.warning(
            "Enabling outlines cache. This is an unbounded on-disk "
            "cache. It may consume a lot of disk space and should "
            "not be used with untrusted clients."
        )
        cache = OutlinesDiskCache(cache_dir)
        outlines_version = importlib.metadata.version("outlines_core")

        cached_version = cache.get("__version__", None)
        if cached_version != outlines_version:
            cache.clear()
        cache.set("__version__", outlines_version)
        return cache

    return LRUCache(maxsize=128)

get_outlines_cache_path()

Get the context object that contains previously-computed return values.

Source code in vllm/v1/structured_output/utils.py

def get_outlines_cache_path() -> str:
    """Get the context object that contains previously-computed return values."""
    outlines_cache_dir = os.getenv("OUTLINES_CACHE_DIR")
    xdg_cache_home = os.getenv("XDG_CACHE_HOME")
    home_dir = os.path.expanduser("~")

    if outlines_cache_dir:
        # OUTLINES_CACHE_DIR takes precedence
        return outlines_cache_dir
    if xdg_cache_home:
        return os.path.join(xdg_cache_home, ".cache", "outlines")
    # If homedir is "/", we may be inside a container, and thus writing to
    # root would be problematic, so we fall back to using a tempfile.
    # Also validate the path exists, since os.path.expanduser does
    # not guarantee existence.
    if os.path.isdir(home_dir) and home_dir != "/":
        # Default Unix fallback: ~/.cache/outlines
        return os.path.join(home_dir, ".cache", "outlines")

    # home_dir may be / inside a docker container without existing user
    tempdir = tempfile.gettempdir()
    return os.path.join(tempdir, ".cache", "outlines")

get_outlines_vocabulary(tokenizer)

Get the Vocabulary object for a given tokenizer.

Source code in vllm/v1/structured_output/utils.py

def get_outlines_vocabulary(tokenizer: TokenizerLike) -> oc.Vocabulary:
    """Get the `Vocabulary` object for a given tokenizer."""
    if hasattr(tokenizer, "_outlines_vocabulary"):
        return tokenizer._outlines_vocabulary  # type: ignore

    reduced_vocab = _reduced_vocabulary(tokenizer)
    vocabulary = OutlinesVocabulary(
        oc.Vocabulary(tokenizer.eos_token_id, reduced_vocab)
    )
    tokenizer._outlines_vocabulary = vocabulary  # type: ignore

    return vocabulary

grammar_is_likely_lark(grammar_str)

Check if grammar appears to use Lark syntax.

Parameters:

• grammar_str

  (str) –

  Input grammar string

Returns:

• bool ( bool ) –

  True if grammar appears to be in Lark format, False otherwise

Examples:

>>> grammar_is_likely_lark("rule: 'abc'")
True
>>> grammar_is_likely_lark("rule ::= 'abc'")
False

Source code in vllm/v1/structured_output/utils.py

def grammar_is_likely_lark(grammar_str: str) -> bool:
    """Check if grammar appears to use Lark syntax.

    Args:
        grammar_str: Input grammar string

    Returns:
        bool: True if grammar appears to be in Lark format, False otherwise

    Examples:
        >>> grammar_is_likely_lark("rule: 'abc'")
        True
        >>> grammar_is_likely_lark("rule ::= 'abc'")
        False

    """
    if not grammar_str or not isinstance(grammar_str, str):
        return False

    for line in grammar_str.split("\n"):
        # Remove both comment styles
        line = re.sub(r"(#|//).*$", "", line).strip()
        if not line:
            continue

        # Look for EBNF rule definition
        if "::=" in line:
            return False

    return True

strip_speculative_padding(token_ids)

Drop speculative-decoding padding from a token block.

ngram and other speculative backends pad rejected draft positions with a -1 sentinel. Structured-output grammars treat every entry as a real token id, so the sentinels (and everything after the first one) are removed here, before tokens reach any backend, rather than inside a single backend.

Source code in vllm/v1/structured_output/utils.py

def strip_speculative_padding(token_ids: list[int]) -> list[int]:
    """Drop speculative-decoding padding from a token block.

    ngram and other speculative backends pad rejected draft positions with a
    -1 sentinel. Structured-output grammars treat every entry as a real token
    id, so the sentinels (and everything after the first one) are removed here,
    before tokens reach any backend, rather than inside a single backend.
    """
    for i, token_id in enumerate(token_ids):
        if token_id < 0:
            return token_ids[:i]
    return token_ids