Skip to content

vllm.v1.structured_output.backend_guidance

logger module-attribute 

logger = init_logger(__name__)

GuidanceBackend dataclass

Bases: StructuredOutputBackend

Source code in vllm/v1/structured_output/backend_guidance.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
@dataclass
class GuidanceBackend(StructuredOutputBackend):
    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
        )

        self.ll_tokenizer = llguidance_hf.from_tokenizer(
            self.tokenizer, self.vocab_size
        )

    def compile_grammar(
        self, request_type: StructuredOutputOptions, grammar_spec: str
    ) -> 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

__init__ 

__init__(
    vllm_config: VllmConfig,
    tokenizer: TokenizerLike,
    vocab_size: int,
) -> None

__post_init__ 

__post_init__()
Source code in vllm/v1/structured_output/backend_guidance.py 
87
88
89
90
91
92
93
94
95
96
97
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
    )

    self.ll_tokenizer = llguidance_hf.from_tokenizer(
        self.tokenizer, self.vocab_size
    )

allocate_token_bitmask 

allocate_token_bitmask(max_num_seqs: int)
Source code in vllm/v1/structured_output/backend_guidance.py 
124
125
126
127
def allocate_token_bitmask(self, max_num_seqs: int):
    return llguidance_torch.allocate_token_bitmask(
        max_num_seqs, self.ll_tokenizer.vocab_size
    )

compile_grammar 

compile_grammar(
    request_type: StructuredOutputOptions, grammar_spec: str
) -> StructuredOutputGrammar
Source code in vllm/v1/structured_output/backend_guidance.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def compile_grammar(
    self, request_type: StructuredOutputOptions, grammar_spec: str
) -> 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

destroy 

destroy()
Source code in vllm/v1/structured_output/backend_guidance.py 
129
130
def destroy(self):
    pass

GuidanceGrammar dataclass

Bases: StructuredOutputGrammar

Source code in vllm/v1/structured_output/backend_guidance.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
@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 []

        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()

ll_matcher instance-attribute 

ll_matcher: LLMatcher

ll_tokenizer instance-attribute 

ll_tokenizer: LLTokenizer

printed_error class-attribute instance-attribute 

printed_error: bool = False

rollback_lag class-attribute instance-attribute 

rollback_lag: int = 0

terminated class-attribute instance-attribute 

terminated: bool = False

vocab_size instance-attribute 

vocab_size: int

__init__ 

__init__(
    ll_matcher: LLMatcher,
    ll_tokenizer: LLTokenizer,
    vocab_size: int,
    printed_error: bool = False,
    terminated: bool = False,
    rollback_lag: int = 0,
) -> None

accept_tokens 

accept_tokens(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.

Source code in vllm/v1/structured_output/backend_guidance.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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

check_error 

check_error()
Source code in vllm/v1/structured_output/backend_guidance.py 
142
143
144
145
146
147
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)

fill_bitmask 

fill_bitmask(bitmask: Tensor, idx: int) -> None
Source code in vllm/v1/structured_output/backend_guidance.py 
201
202
203
204
205
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()

is_terminated 

is_terminated() -> bool
Source code in vllm/v1/structured_output/backend_guidance.py 
207
208
def is_terminated(self) -> bool:
    return self.terminated

reset 

reset()
Source code in vllm/v1/structured_output/backend_guidance.py 
210
211
212
def reset(self):
    # This method may be not needed anymore? TODO
    self.ll_matcher.reset()

rollback 

rollback(num_tokens: int) -> None
Source code in vllm/v1/structured_output/backend_guidance.py 
194
195
196
197
198
199
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()

validate_tokens 

validate_tokens(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.

Source code in vllm/v1/structured_output/backend_guidance.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
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 []

    num_tokens = self.ll_matcher.validate_tokens(tokens)

    self.check_error()

    return tokens[:num_tokens]

_walk_json_for_additional_properties 

_walk_json_for_additional_properties(data: object)
Source code in vllm/v1/structured_output/backend_guidance.py 
34
35
36
37
38
39
40
41
42
43
44
def _walk_json_for_additional_properties(data: object):
    if isinstance(data, dict):
        for value in data.values():
            _walk_json_for_additional_properties(value)
        if "additionalProperties" not in data and (
            "properties" in data or "patternProperties" in data
        ):
            data["additionalProperties"] = False
    elif isinstance(data, list):
        for item in data:
            _walk_json_for_additional_properties(item)

has_guidance_unsupported_json_features 

has_guidance_unsupported_json_features(
    schema: dict[str, Any],
) -> bool

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

Source code in vllm/v1/structured_output/backend_guidance.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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)

process_for_additional_properties 

process_for_additional_properties(
    guide_json: str | dict[str, Any],
) -> dict[str, Any]
Source code in vllm/v1/structured_output/backend_guidance.py 
73
74
75
76
77
78
79
80
81
82
def process_for_additional_properties(
    guide_json: str | dict[str, Any],
) -> dict[str, Any]:
    if isinstance(guide_json, str):
        guide_json_obj = json.loads(guide_json)
    else:
        # copy for modifications
        guide_json_obj = copy.deepcopy(guide_json)
    _walk_json_for_additional_properties(guide_json_obj)
    return guide_json_obj

serialize_guidance_grammar 

serialize_guidance_grammar(
    request_type: StructuredOutputOptions,
    grammar_spec: str | dict[str, Any],
    disable_any_whitespace: bool = False,
    disable_additional_properties: bool = False,
) -> str
Source code in vllm/v1/structured_output/backend_guidance.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def serialize_guidance_grammar(
    request_type: StructuredOutputOptions,
    grammar_spec: str | dict[str, Any],
    disable_any_whitespace: bool = False,
    disable_additional_properties: bool = False,
) -> str:
    def _process_schema(
        grammar_spec: str | dict[str, Any],
    ) -> str:
        if disable_additional_properties:
            grammar_spec = process_for_additional_properties(grammar_spec)
        return llguidance.LLMatcher.grammar_from_json_schema(
            grammar_spec,
            defaults={
                "whitespace_flexible": not disable_any_whitespace,
            },
        )

    if request_type == StructuredOutputOptions.JSON:
        return _process_schema(grammar_spec)
    elif request_type == StructuredOutputOptions.JSON_OBJECT:
        return llguidance.LLMatcher.grammar_from_json_schema(
            '{"type": "object"}',
            defaults={
                "whitespace_flexible": not disable_any_whitespace,
            },
        )
    else:
        if request_type == StructuredOutputOptions.REGEX:
            tp = "regex"
        elif request_type == StructuredOutputOptions.GRAMMAR:
            tp = "grammar"
        elif request_type == StructuredOutputOptions.CHOICE:
            tp = "choice"
        elif request_type == StructuredOutputOptions.STRUCTURAL_TAG:
            if isinstance(grammar_spec, str):
                s_tag = json.loads(grammar_spec)
            else:
                s_tag = grammar_spec
            triggers: list[str] = s_tag["triggers"]
            tags: list[llguidance.StructTag] = []
            for s in s_tag["structures"]:
                begin: str = s["begin"]
                trig = next((t for t in triggers if begin.startswith(t)), None)
                if trig is None:
                    raise ValueError(
                        f"Trigger {begin} not found in triggers {triggers}"
                    )
                tags.append(
                    llguidance.StructTag(
                        trigger=trig,
                        begin=s["begin"],
                        grammar=_process_schema(s["schema"]),
                        end=s["end"],
                    )
                )
            if not tags:
                raise ValueError("No structural tags found in the grammar spec.")
            return llguidance.StructTag.to_grammar(tags)
        else:
            logger.error(
                "Validation should have already occurred. Please file an issue."
            )
            raise ValueError(
                f"grammar is not of valid supported types. ({request_type!s})"
            )
        return llguidance.grammar_from(tp, grammar_spec)

validate_guidance_grammar 

validate_guidance_grammar(
    sampling_params: SamplingParams,
    tokenizer: LLTokenizer | None = None,
) -> None
Source code in vllm/v1/structured_output/backend_guidance.py 
284
285
286
287
288
289
290
291
def validate_guidance_grammar(
    sampling_params: SamplingParams, tokenizer: llguidance.LLTokenizer | None = None
) -> None:
    tp, grm = get_structured_output_key(sampling_params.structured_outputs)
    guidance_grm = serialize_guidance_grammar(tp, grm)
    err = llguidance.LLMatcher.validate_grammar(guidance_grm, tokenizer)
    if err:
        raise ValueError(f"Grammar error: {err}")