vllm.v1.structured_output.backend_guidance

Classes:

• GuidanceBackend –
• GuidanceGrammar –

Functions:

• has_guidance_unsupported_json_features –

  Check if JSON schema contains features unsupported by guidance/llguidance.

GuidanceBackend dataclass

Bases: StructuredOutputBackend

Methods:

• is_tokenizer_supported –

  Check if tokenizer is supported by guidance backend.

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

@dataclass
class GuidanceBackend(StructuredOutputBackend):
    @staticmethod
    def is_tokenizer_supported(tokenizer: TokenizerLike) -> bool:
        """Check if tokenizer is supported by guidance backend.

        TODO(arpera):
        guidance backend does NOT support some types of tokenizers
        so, we MUST verify tokenizer before using guidance
        The problem is that guidance API does not provide
        a way to validate tokenizer before building LLTokenizer
        So, we have to do it manually here in vLLM.
        This is ofcourse very bad, we duplicate complex logic
        that can change arbitrary in future.
        Our check here is based on guidance source code
        on function `from_tokenizer` implementation:
        https://github.com/guidance-ai/llguidance/blob/main/python/llguidance/hf.py#L29-L47
        If you want to contribute to vLLM, please, add to llguidance
        API function to implement this check there.
        In that case we will remove this check from vLLM.
        That kind of work is really appreciated!
        """
        if is_mistral_tokenizer(tokenizer):
            return tokenizer.is_tekken
        # PreTrainedTokenizerFast is the same as TokenizersBackend
        return isinstance(tokenizer, TokenizersBackend)

    @staticmethod
    def _build_ll_tokenizer(tokenizer: TokenizerLike, vocab_size: int) -> Any:
        if is_mistral_tokenizer(tokenizer):
            return tokenizer.llg_tokenizer
        return llguidance_hf.from_tokenizer(tokenizer, max(vocab_size, len(tokenizer)))

    def __post_init__(self):
        self.disable_any_whitespace = (
            self.vllm_config.structured_outputs_config.disable_any_whitespace
        )
        self.disable_additional_properties = (
            self.vllm_config.structured_outputs_config.disable_additional_properties
        )

        if not self.is_tokenizer_supported(self.tokenizer):
            raise ValueError(
                "Only fast tokenizers are supported."
                "This error should NOT happen,"
                "if you see this, please report an issue."
            )
        self.ll_tokenizer = self._build_ll_tokenizer(self.tokenizer, self.vocab_size)

    def compile_grammar(
        self,
        request_type: StructuredOutputOptions,
        grammar_spec: str,
        stop_token_ids: set[int] | None = None,
    ) -> StructuredOutputGrammar:
        self.serialized_grammar = serialize_guidance_grammar(
            request_type,
            grammar_spec,
            self.disable_any_whitespace,
            self.disable_additional_properties,
        )

        ll_matcher = llguidance.LLMatcher(
            self.ll_tokenizer,
            self.serialized_grammar,
            log_level=int(os.environ.get("LLGUIDANCE_LOG_LEVEL", "1")),
        )

        if err := ll_matcher.get_error():
            raise ValueError(f"Grammar error: {err}")

        return GuidanceGrammar(
            ll_matcher=ll_matcher,
            ll_tokenizer=self.ll_tokenizer,
            vocab_size=self.vocab_size,
        )

    def allocate_token_bitmask(self, max_num_seqs: int):
        return llguidance_torch.allocate_token_bitmask(
            max_num_seqs, self.ll_tokenizer.vocab_size
        )

    def destroy(self):
        pass

is_tokenizer_supported(tokenizer) staticmethod

Check if tokenizer is supported by guidance backend.

TODO(arpera): guidance backend does NOT support some types of tokenizers so, we MUST verify tokenizer before using guidance The problem is that guidance API does not provide a way to validate tokenizer before building LLTokenizer So, we have to do it manually here in vLLM. This is ofcourse very bad, we duplicate complex logic that can change arbitrary in future. Our check here is based on guidance source code on function from_tokenizer implementation: https://github.com/guidance-ai/llguidance/blob/main/python/llguidance/hf.py#L29-L47 If you want to contribute to vLLM, please, add to llguidance API function to implement this check there. In that case we will remove this check from vLLM. That kind of work is really appreciated!

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

@staticmethod
def is_tokenizer_supported(tokenizer: TokenizerLike) -> bool:
    """Check if tokenizer is supported by guidance backend.

    TODO(arpera):
    guidance backend does NOT support some types of tokenizers
    so, we MUST verify tokenizer before using guidance
    The problem is that guidance API does not provide
    a way to validate tokenizer before building LLTokenizer
    So, we have to do it manually here in vLLM.
    This is ofcourse very bad, we duplicate complex logic
    that can change arbitrary in future.
    Our check here is based on guidance source code
    on function `from_tokenizer` implementation:
    https://github.com/guidance-ai/llguidance/blob/main/python/llguidance/hf.py#L29-L47
    If you want to contribute to vLLM, please, add to llguidance
    API function to implement this check there.
    In that case we will remove this check from vLLM.
    That kind of work is really appreciated!
    """
    if is_mistral_tokenizer(tokenizer):
        return tokenizer.is_tekken
    # PreTrainedTokenizerFast is the same as TokenizersBackend
    return isinstance(tokenizer, TokenizersBackend)

GuidanceGrammar dataclass

Bases: StructuredOutputGrammar

Methods:

• accept_tokens –

  Accepts a list of tokens and advances the parser.

• validate_tokens –

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

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

@dataclass
class GuidanceGrammar(StructuredOutputGrammar):
    ll_matcher: llguidance.LLMatcher
    ll_tokenizer: llguidance.LLTokenizer
    vocab_size: int
    printed_error: bool = False
    terminated: bool = False
    rollback_lag: int = 0

    def check_error(self):
        if not self.printed_error:
            err = self.ll_matcher.get_error()
            if err:
                self.printed_error = True
                logger.warning("LLMatcher error: %s", err)

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

        Returns True if the parser was advanced successfully.
        Returns False if the parser failed to advance.
        """
        eos_token = self.ll_tokenizer.eos_token
        eos_index = next(
            (index for index, token in enumerate(tokens) if token == eos_token),
            None,
        )
        if eos_index is not None and eos_index != len(tokens) - 1:
            return False

        if self.terminated:
            return True

        # TODO - Add jump decoding support in the future:
        # self.ll_matcher.compute_ff_bytes() - this should always work
        # self.ll_matcher.compute_ff_tokens() - this only works for
        #   "canonical" tokenizers
        # For conversion between the two, see
        # https://github.com/guidance-ai/llguidance/blob/main/docs/fast_forward.md

        num_tokens = self.ll_matcher.try_consume_tokens(tokens)
        if num_tokens != len(tokens):
            if (
                eos_index is not None
                and num_tokens == eos_index
                and self.ll_matcher.is_stopped()
                and self.ll_matcher.is_accepting()
            ):
                # LLMatcher stops at an accepting grammar state and leaves the
                # wrapper-level EOS token unconsumed.
                self.rollback_lag = 1
                self.terminated = True
                self.check_error()
                return True
            if num_tokens > 0:
                self.ll_matcher.rollback(num_tokens)
            self.check_error()
            return False

        if eos_index is not None:
            self.terminated = True

        self.check_error()

        return True

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

        Returns the prefix list of tokens that are accepted by the parser.
        """
        if len(tokens) == 0:
            return []
        if self.ll_matcher.is_stopped():
            return []

        tokens = strip_speculative_padding(tokens)
        if len(tokens) == 0:
            return []

        num_tokens = self.ll_matcher.validate_tokens(tokens)

        self.check_error()

        return tokens[:num_tokens]

    def rollback(self, num_tokens: int) -> None:
        if num_tokens > 0:
            self.ll_matcher.rollback(num_tokens - self.rollback_lag)
            self.terminated = False
            self.rollback_lag = 0
            self.check_error()

    def fill_bitmask(self, bitmask: torch.Tensor, idx: int) -> None:
        # this will automatically return [EOS] mask if the matcher is stopped
        # or otherwise in an error state
        llguidance_torch.fill_next_token_bitmask(self.ll_matcher, bitmask, idx)
        self.check_error()

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

    def reset(self):
        # This method may be not needed anymore? TODO
        self.ll_matcher.reset()

accept_tokens(request_id, tokens)

Accepts a list of tokens and advances the parser.

Returns True if the parser was advanced successfully. Returns False if the parser failed to advance.

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

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

    Returns True if the parser was advanced successfully.
    Returns False if the parser failed to advance.
    """
    eos_token = self.ll_tokenizer.eos_token
    eos_index = next(
        (index for index, token in enumerate(tokens) if token == eos_token),
        None,
    )
    if eos_index is not None and eos_index != len(tokens) - 1:
        return False

    if self.terminated:
        return True

    # TODO - Add jump decoding support in the future:
    # self.ll_matcher.compute_ff_bytes() - this should always work
    # self.ll_matcher.compute_ff_tokens() - this only works for
    #   "canonical" tokenizers
    # For conversion between the two, see
    # https://github.com/guidance-ai/llguidance/blob/main/docs/fast_forward.md

    num_tokens = self.ll_matcher.try_consume_tokens(tokens)
    if num_tokens != len(tokens):
        if (
            eos_index is not None
            and num_tokens == eos_index
            and self.ll_matcher.is_stopped()
            and self.ll_matcher.is_accepting()
        ):
            # LLMatcher stops at an accepting grammar state and leaves the
            # wrapper-level EOS token unconsumed.
            self.rollback_lag = 1
            self.terminated = True
            self.check_error()
            return True
        if num_tokens > 0:
            self.ll_matcher.rollback(num_tokens)
        self.check_error()
        return False

    if eos_index is not None:
        self.terminated = True

    self.check_error()

    return True

validate_tokens(tokens)

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

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

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

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

    Returns the prefix list of tokens that are accepted by the parser.
    """
    if len(tokens) == 0:
        return []
    if self.ll_matcher.is_stopped():
        return []

    tokens = strip_speculative_padding(tokens)
    if len(tokens) == 0:
        return []

    num_tokens = self.ll_matcher.validate_tokens(tokens)

    self.check_error()

    return tokens[:num_tokens]

has_guidance_unsupported_json_features(schema)

Check if JSON schema contains features unsupported by guidance/llguidance.

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

def has_guidance_unsupported_json_features(schema: dict[str, Any]) -> bool:
    """Check if JSON schema contains features unsupported by guidance/llguidance."""

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

        # patternProperties is not supported by llguidance
        if "patternProperties" in obj:
            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)