vllm.v1.structured_output.backend_xgrammar

Classes:

• XgrammarGrammar –

Functions:

• has_xgrammar_unsupported_json_features –

  Check if JSON schema contains features unsupported by xgrammar.

• validate_xgrammar_grammar –

  Validate that the request is supported by structured output.

XgrammarGrammar dataclass

Bases: StructuredOutputGrammar

Methods:

• accept_tokens –

  Accepts a list of tokens and advances the FSM.

• validate_tokens –

  Checks if the list of tokens are accepted by the FSM in sequence.

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

@dataclass
class XgrammarGrammar(StructuredOutputGrammar):
    # NOTE: This would be a generic-enough class for
    # supporting different backends, in the future.
    # For now, just xgrammar.
    #
    # https://xgrammar.mlc.ai/docs/api/python/index.html#xgrammar.GrammarMatcher.find_jump_forward_string
    # for jump-forward decoding

    vocab_size: int
    matcher: xgr.GrammarMatcher = field(hash=False)
    ctx: xgr.CompiledGrammar = field(hash=False)
    num_processed_tokens: int = field(
        default_factory=lambda: 0, repr=False, hash=False, init=False
    )
    _is_terminated: bool = field(default=False, repr=False, hash=False)

    def accept_tokens(self, request_id: str, tokens: list[int]) -> bool:
        """Accepts a list of tokens and advances the FSM.

        Returns True if all grammar-constrained tokens were accepted.
        Tokens after termination are ignored. Returns False if the FSM
        failed to advance.
        """
        if self._is_terminated:
            return True
        for token in tokens:
            if not self.matcher.accept_token(token):
                logger.error(
                    "Failed to advance FSM for request %s "
                    "for tokens %s. Please file an issue.",
                    request_id,
                    token,
                )
                return False
            self.num_processed_tokens += 1
            self._is_terminated = self.matcher.is_terminated()
            if self._is_terminated:
                break
        return True

    def validate_tokens(self, tokens: list[int]) -> list[int]:
        """Checks if the list of tokens are accepted by the FSM in sequence.
        Will not advance the FSM.

        Returns the prefix list of tokens that are accepted by the FSM.
        """
        if self._is_terminated:
            return []

        accepted_tokens = []
        for token in tokens:
            if self.matcher.accept_token(token):
                accepted_tokens.append(token)
                if self.matcher.is_terminated():
                    break
            else:
                break
        if len(accepted_tokens) > 0:
            # Rollback the FSM to the initial state
            self.matcher.rollback(len(accepted_tokens))
        return accepted_tokens

    def rollback(self, num_tokens: int) -> None:
        self.matcher.rollback(num_tokens)
        self.num_processed_tokens -= num_tokens
        self._is_terminated = self.matcher.is_terminated()

    def fill_bitmask(self, bitmask: torch.Tensor, idx: int) -> None:
        self.matcher.fill_next_token_bitmask(bitmask, idx)

    def is_terminated(self) -> bool:
        return self._is_terminated

    def reset(self):
        self.matcher.reset()
        self.num_processed_tokens = 0
        self._is_terminated = False

accept_tokens(request_id, tokens)

Accepts a list of tokens and advances the FSM.

Returns True if all grammar-constrained tokens were accepted. Tokens after termination are ignored. Returns False if the FSM failed to advance.

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

def accept_tokens(self, request_id: str, tokens: list[int]) -> bool:
    """Accepts a list of tokens and advances the FSM.

    Returns True if all grammar-constrained tokens were accepted.
    Tokens after termination are ignored. Returns False if the FSM
    failed to advance.
    """
    if self._is_terminated:
        return True
    for token in tokens:
        if not self.matcher.accept_token(token):
            logger.error(
                "Failed to advance FSM for request %s "
                "for tokens %s. Please file an issue.",
                request_id,
                token,
            )
            return False
        self.num_processed_tokens += 1
        self._is_terminated = self.matcher.is_terminated()
        if self._is_terminated:
            break
    return True

validate_tokens(tokens)

Checks if the list of tokens are accepted by the FSM in sequence. Will not advance the FSM.

Returns the prefix list of tokens that are accepted by the FSM.

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

def validate_tokens(self, tokens: list[int]) -> list[int]:
    """Checks if the list of tokens are accepted by the FSM in sequence.
    Will not advance the FSM.

    Returns the prefix list of tokens that are accepted by the FSM.
    """
    if self._is_terminated:
        return []

    accepted_tokens = []
    for token in tokens:
        if self.matcher.accept_token(token):
            accepted_tokens.append(token)
            if self.matcher.is_terminated():
                break
        else:
            break
    if len(accepted_tokens) > 0:
        # Rollback the FSM to the initial state
        self.matcher.rollback(len(accepted_tokens))
    return accepted_tokens

_schema_types(schema)

Normalize a scalar or list-valued JSON Schema type.

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

def _schema_types(schema: dict[str, Any]) -> set[str]:
    """Normalize a scalar or list-valued JSON Schema type."""
    schema_type = schema.get("type")
    if isinstance(schema_type, str):
        return {schema_type}
    if isinstance(schema_type, list):
        return {item for item in schema_type if isinstance(item, str)}
    return set()

has_xgrammar_unsupported_json_features(schema)

Check if JSON schema contains features unsupported by xgrammar.

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

def has_xgrammar_unsupported_json_features(schema: dict[str, Any]) -> bool:
    """Check if JSON schema contains features unsupported by xgrammar."""

    def check_object(obj: dict[str, Any]) -> bool:
        if not isinstance(obj, dict):
            return False

        schema_types = _schema_types(obj)

        # Check for numeric ranges
        if (schema_types & {"integer", "number"}) and ("multipleOf" in obj):
            return True

        # Check for array unsupported keywords
        if "array" in schema_types and any(
            key in obj
            for key in ("uniqueItems", "contains", "minContains", "maxContains")
        ):
            return True

        # Unsupported keywords for strings
        if (
            "string" in schema_types
            and "format" in obj
            and obj["format"] not in STRING_SUPPORTED_FORMATS
        ):
            return True

        # A string mixing a generative constraint (pattern or format) with
        # explicit length bounds. xgrammar compiles the pattern/format side
        # and silently drops minLength/maxLength from the grammar, so output
        # can violate the bound without any error surfacing. Verified against
        # the compiled EBNF: pattern/format grammars come out byte-identical
        # with and without the length keywords, while maxLength alone lowers
        # to {0, N} correctly.
        if "string" in schema_types and _has_pattern_and_length_bounds(obj):
            return True

        # propertyNames validates names, so it is a string schema even when it
        # omits "type", which is the form that escapes the check above.
        if (
            "object" in schema_types
            and isinstance(obj.get("propertyNames"), dict)
            and _has_pattern_and_length_bounds(obj["propertyNames"])
        ):
            return True

        # FIXME: propertyNames conflicts with properties/patternProperties/
        # additionalProperties/unevaluatedProperties under xgrammar.
        # https://github.com/mlc-ai/xgrammar/issues/826
        if (
            "object" in schema_types
            and "propertyNames" in obj
            and (
                "properties" in obj
                or "patternProperties" in obj
                or isinstance(obj.get("additionalProperties"), dict)
                or obj.get("unevaluatedProperties", True) is not True
            )
        ):
            return True

        # FIXME: multiple patternProperties, or patternProperties alongside
        # properties, conflict under xgrammar.
        if (
            "object" in schema_types
            and isinstance(obj.get("patternProperties"), dict)
            and ("properties" in obj or len(obj["patternProperties"]) > 1)
        ):
            return True

        # Recursively check all nested objects and arrays
        for value in obj.values():
            if isinstance(value, dict):
                if check_object(value):
                    return True
            elif isinstance(value, list):
                for item in value:
                    if isinstance(item, dict) and check_object(item):
                        return True

        return False

    return check_object(schema)

validate_xgrammar_grammar(sampling_params)

Validate that the request is supported by structured output.

Raises VLLMValidationError if the request is not supported.

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

def validate_xgrammar_grammar(sampling_params: SamplingParams) -> None:
    """Validate that the request is supported by structured output.

    Raises VLLMValidationError if the request is not supported.
    """
    if sampling_params.structured_outputs is None:
        return

    so_params = sampling_params.structured_outputs

    if so_params.regex:
        # A NUL byte is never meaningful in a regex pattern and is not handled
        # by xgrammar's native regex converter. Reject it here, before the
        # pattern reaches that native code; the try/except below does not cover
        # this case.
        if "\x00" in so_params.regex:
            raise ValueError(
                "structured_outputs.regex must not contain a NUL character ('\\x00')"
            )
        try:
            compile_regex_with_timeout(
                xgr.Grammar.from_regex,
                so_params.regex,
            )
        except Exception as err:
            raise VLLMValidationError(
                f"Failed to transform regex into a grammar: {err}"
            ) from err

    if so_params.choice:
        choice_grammar = choice_as_grammar(so_params.choice)
        try:
            xgr.Grammar.from_ebnf(choice_grammar)
        except Exception as err:
            raise VLLMValidationError(
                f"Failed to transform choices into a grammar: {err}"
            ) from err
        so_params.choice = None
        so_params.grammar = choice_grammar
        return

    if so_params.json:
        if isinstance(so_params.json, str):
            try:
                schema = json.loads(so_params.json)
            except json.JSONDecodeError as e:
                raise VLLMValidationError("Invalid JSON grammar specification.") from e
        else:
            schema = so_params.json

        if has_xgrammar_unsupported_json_features(schema):
            raise VLLMValidationError(
                "The provided JSON schema contains features not supported by xgrammar."
            )

        try:
            xgr.Grammar.from_json_schema(schema)
        except Exception as err:
            raise VLLMValidationError(
                f"Failed to transform json schema into a grammar: {err}"
            ) from err
        return

    if so_params.grammar:
        # Parse the grammar with the same syntax `compile_grammar` will use,
        # but don't compile it. The grammar is passed on unchanged.
        try:
            if grammar_is_likely_lark(so_params.grammar):
                xgr.Grammar.from_lark(so_params.grammar)
            else:
                xgr.Grammar.from_ebnf(so_params.grammar)
        except Exception as e:
            raise VLLMValidationError("Invalid grammar specification.") from e
        return

    if so_params.structural_tag:
        try:
            s_tag = json.loads(so_params.structural_tag)

            # Using the deprecated method of compiling structural tag
            if "structures" in s_tag:
                tags = [
                    xgr.StructuralTagItem(
                        begin=s["begin"],
                        schema=json.dumps(s["schema"]),
                        end=s["end"],
                    )
                    for s in s_tag["structures"]
                ]
                xgr.Grammar.from_structural_tag(tags, s_tag["triggers"])
            else:
                xgr.Grammar.from_structural_tag(so_params.structural_tag)
        except Exception as e:
            raise VLLMValidationError("Invalid structural tag specification.") from e