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")),
        )

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

        r.check_error()
        return r

    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.
        """
        if self.ll_tokenizer.eos_token in tokens:
            if self.ll_matcher.is_stopped() and not self.terminated:
                self.rollback_lag = 1
            self.terminated = True

        if self.ll_matcher.is_stopped():
            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

        r = self.ll_matcher.consume_tokens(tokens)

        self.check_error()

        return r

    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.
    """
    if self.ll_tokenizer.eos_token in tokens:
        if self.ll_matcher.is_stopped() and not self.terminated:
            self.rollback_lag = 1
        self.terminated = True

    if self.ll_matcher.is_stopped():
        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

    r = self.ll_matcher.consume_tokens(tokens)

    self.check_error()

    return r

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)