vllm.parser.response_template_grammar

Structural tags for tool calls and answers, derived from a response template.

Functions:

• answer_rest_structural_tag –

  The rest of an answer that is already open, or None if underivable.

• message_start_anchors –

  The strings the assistant message starts after, if literal.

• tool_call_structural_tag –

  Derive native calls to functions in the format of the tool call

_closers(field)

Return literal closers, an empty list if absent, or None if underivable.

Source code in vllm/parser/response_template_grammar.py

def _closers(field: ResponseTemplateField) -> list[str] | None:
    """Return literal closers, an empty list if absent, or `None` if underivable."""
    if field.close_re is None:
        return []
    return field.close_literals or _literals(field.close_re.pattern)

_continuing(literals, anchors, opener)

Keep anchor continuations only where the parser finds the opener.

Source code in vllm/parser/response_template_grammar.py

def _continuing(
    literals: list[str] | None, anchors: list[str] | None, opener: re.Pattern
) -> list[str] | None:
    """Keep anchor continuations only where the parser finds the opener."""
    if not literals or not anchors:
        return literals
    rests = [
        literal[len(anchor) :]
        for literal in literals
        for anchor in anchors
        if len(literal) > len(anchor) and literal.startswith(anchor)
    ]
    rests = [
        rest
        for rest in rests
        if any(opener.fullmatch(rest, index) for index in range(len(rest)))
    ]
    return list(dict.fromkeys([*literals, *rests]))

_json_format(schema, closers)

Use JSON unless a closer is ambiguous with its syntax or the schema is unsupported. Strings hold no closer, as the parser ends the call there.

Source code in vllm/parser/response_template_grammar.py

def _json_format(schema: Any, closers: list[str]) -> dict[str, Any] | None:
    """Use JSON unless a closer is ambiguous with its syntax or the schema is
    unsupported. Strings hold no closer, as the parser ends the call there."""
    if closers and (
        any(set(closer) <= _JSON_SYNTAX for closer in closers)
        or not _json_compiles(json.dumps(schema))
    ):
        return None
    return {"type": "json_schema", "json_schema": schema, "excludes": closers}

_literals(pattern, *, message_start=False, lookahead=False)

Expand literal alternatives, allowing ^ only at the message start and a final lookahead only with lookahead.

Source code in vllm/parser/response_template_grammar.py

def _literals(
    pattern: str, *, message_start: bool = False, lookahead: bool = False
) -> list[str] | None:
    """Expand literal alternatives, allowing `^` only at the message start and
    a final lookahead only with `lookahead`."""
    literals, index = _alternatives(pattern, 0)
    if literals is None or index != len(pattern):
        return None
    if any(literal.count(_AHEAD) > lookahead for literal in literals):
        return None
    if not message_start:
        return None if any(_START in literal for literal in literals) else literals
    return [
        literal.removeprefix(_START)
        for literal in literals
        if _START not in literal[1:] and literal != _START
    ]

_may_write(pattern, chars)

Whether a match of pattern may hold one of chars, as any pattern may that Python cannot parse.

Source code in vllm/parser/response_template_grammar.py

def _may_write(pattern: str, chars: set[str]) -> bool:
    """Whether a match of `pattern` may hold one of `chars`, as any pattern may
    that Python cannot parse."""
    try:
        parsed = _sre_parse.parse(pattern)
    except (_sre_parse.error, OverflowError, RecursionError):
        return True
    if parsed.state.flags & _sre_parse.SRE_FLAG_IGNORECASE:
        return True

    def may(items: Any) -> bool:
        for op, arg in items:
            if op is _sre_parse.LITERAL:
                written = {chr(arg)}
            elif op is _sre_parse.IN:
                written = {char for char in chars if _in_class(arg, char)}
            elif op in _REPEATS:
                written = chars if arg[1] and may(arg[2]) else set()
            elif op is _sre_parse.SUBPATTERN:
                if arg[1] & _sre_parse.SRE_FLAG_IGNORECASE:
                    return True
                written = chars if may(arg[3]) else set()
            elif op is _sre_parse.BRANCH:
                written = chars if any(map(may, arg[1])) else set()
            elif op is _sre_parse.AT:
                continue
            else:
                # Such as `.`, which XGrammar lets match line breaks.
                return True
            if written & chars:
                return True
        return False

    return may(parsed)

_nonempty_start(closers)

The start of text that is not empty and does not begin with one of the closers. Each branch leaves every closer's spelling before completing it, on a character that starts no closer, so the text after it cannot hide a closer's start and a closer written as one token cannot pass for the text.

Source code in vllm/parser/response_template_grammar.py

def _nonempty_start(closers: list[str]) -> dict[str, Any]:
    """The start of text that is not empty and does not begin with one of the
    `closers`. Each branch leaves every closer's spelling before completing it,
    on a character that starts no closer, so the text after it cannot hide a
    closer's start and a closer written as one token cannot pass for the text."""
    closers = [closer for closer in closers if closer]
    starts = {closer[0] for closer in closers}
    prefixes = dict.fromkeys(
        [""] + [closer[:end] for closer in closers for end in range(1, len(closer))]
    )
    branches = []
    for prefix in prefixes:
        # A prefix holding a whole closer has already closed the text.
        if any(closer in prefix for closer in closers):
            continue
        nexts = {
            closer[len(prefix)]
            for closer in closers
            if len(closer) > len(prefix) and closer.startswith(prefix)
        }
        excluded = "".join(map(_class_escape, sorted(starts | nexts)))
        branches.append(f"{_escape(prefix)}[^{excluded}]")
    return {"type": "regex", "pattern": "|".join(branches)}

_openers(field, message_start=None)

Derive field openers, including continuations of message-start anchors.

Source code in vllm/parser/response_template_grammar.py

def _openers(
    field: ResponseTemplateField, message_start: list[str] | None = None
) -> list[str] | None:
    """Derive field openers, including continuations of message-start anchors."""
    if field.open_literals or field.open_re is None:
        openers = field.open_literals
    else:
        openers = _literals(
            field.open_re.pattern, message_start=message_start is not None
        )
    return _continuing(openers, message_start, field.open_re)

_other_names(pattern, names)

A regex for the names pattern matches besides names, or None if pattern is no negatable character class repeated.

Source code in vllm/parser/response_template_grammar.py

def _other_names(pattern: str, names: list[str]) -> str | None:
    """A regex for the names `pattern` matches besides `names`, or `None` if
    `pattern` is no negatable character class repeated."""
    if not names:
        return pattern
    match = re.fullmatch(r"\\([wS])\+|\[\^((?:\\.|[^\]\\])+)\]\+", pattern)
    if match is None:
        return None
    negated = {"w": r"\W", "S": r"\s"}[match[1]] if match[1] else match[2]

    def other(chars: set[str]) -> str:
        return f"[^{negated}{''.join(map(_class_escape, sorted(chars)))}]"

    trie: dict[str, dict] = {}
    for name in names:
        node = trie
        for char in name:
            node = node.setdefault(char, {})
        node[""] = {}

    def rest(node: dict[str, dict], first: bool) -> str:
        """The rest of a name past `node` that ends on none of `names`."""
        branches = [other(set(node) - {""}) + other(set()) + "*"]
        branches += [
            _escape(char) + rest(child, False) for char, child in node.items() if char
        ]
        return f"(?:{'|'.join(branches)})" + ("" if first or "" in node else "?")

    return rest(trie, True)

_reply_format(answer, message_start, content)

The answer field holding content, or free text if not given, or None if it cannot be derived.

Source code in vllm/parser/response_template_grammar.py

def _reply_format(
    answer: ResponseTemplateField,
    message_start: list[str] | None,
    content: dict[str, Any] | None,
) -> dict[str, Any] | None:
    """The `answer` field holding `content`, or free text if not given, or
    `None` if it cannot be derived."""
    openers, closers = _openers(answer, message_start), _closers(answer)
    if not openers or closers is None:
        return None
    return _sequence(_const_strings(openers), *_answer_text(closers, content))

_spells_closer(closers, pieces, *, repeats=False) cached

Check whether argument framing can spell a call closer across pieces.

Source code in vllm/parser/response_template_grammar.py

@lru_cache(maxsize=512)
def _spells_closer(
    closers: tuple[str, ...],
    pieces: tuple[tuple[str, str, bool], ...],
    *,
    repeats: bool = False,
) -> bool:
    """Check whether argument framing can spell a call closer across pieces."""
    texts = [text for text, _, _ in pieces]
    loop = f"(?:{''.join(texts)})*" if repeats else ""
    # Another call or an answer may follow the call's own closer.
    closed = "(?:" + "|".join(map(re.escape, closers)) + r")[\s\S]*"
    for index, (_, endings, holds) in enumerate(pieces):
        ends = re.compile(endings)
        rest = re.compile("".join(texts[index + 1 :]) + loop + closed, re.DOTALL)
        for closer in closers:
            if holds and ends.fullmatch(closer, partial=True):
                return True
            # A closer that starts in this piece and ends in the text after it.
            for split in range(1, len(closer)):
                if ends.fullmatch(closer[:split]) and rest.fullmatch(
                    closer[split:], partial=True
                ):
                    return True
    return False

_split_group(pattern, name)

Split a named regex group into its prefix, body and suffix.

Source code in vllm/parser/response_template_grammar.py

def _split_group(pattern: str, name: str) -> tuple[str, str, str] | None:
    """Split a named regex group into its prefix, body and suffix."""
    opener = f"(?P<{name}>"
    start = pattern.find(opener)
    if start < 0:
        return None
    depth = 0
    in_class = False
    index = start
    while index < len(pattern):
        char = pattern[index]
        if char == "\\":
            index += 2
            continue
        if in_class:
            in_class = char != "]"
        elif char == "[":
            in_class = True
        elif char == "(":
            depth += 1
        elif char == ")":
            depth -= 1
            if depth == 0:
                body = pattern[start + len(opener) : index]
                return pattern[:start], body, pattern[index + 1 :]
        index += 1
    return None

_string_regex(schema, excludes)

Hold the format or pattern of a raw string. Like XGrammar, a format overrides a pattern, either one drops length bounds, and the pattern matches the whole string.

Source code in vllm/parser/response_template_grammar.py

def _string_regex(schema: dict[str, Any], excludes: list[str]) -> dict[str, Any] | None:
    """Hold the `format` or `pattern` of a raw string. Like XGrammar, a format
    overrides a pattern, either one drops length bounds, and the pattern
    matches the whole string."""
    if set(schema) - _ANNOTATIONS - {
        "type",
        "format",
        "pattern",
        "minLength",
        "maxLength",
    } or _schema_types(schema) != ("string",):
        return None
    pattern = (
        _STRING_FORMATS.get(schema["format"])
        if "format" in schema
        else schema["pattern"]
    )
    if (
        not isinstance(pattern, str)
        or _may_write(pattern, {text[0] for text in excludes if text})
        or not _regex_compiles(pattern)
    ):
        return None
    return {"type": "regex", "pattern": pattern}

_structural_tag(output, thinking, message_start, requires_reasoning=False)

Frame output after reasoning, from the first token at the message_start. The reasoning is optional unless requires_reasoning makes the message open with non-empty reasoning.

Source code in vllm/parser/response_template_grammar.py

def _structural_tag(
    output: Callable[[list[str] | None], list[dict[str, Any]] | None],
    thinking: ResponseTemplateField | None,
    message_start: list[str] | None,
    requires_reasoning: bool = False,
) -> dict[str, Any] | None:
    """Frame output after reasoning, from the first token at the
    `message_start`. The reasoning is optional unless `requires_reasoning`
    makes the message open with non-empty reasoning."""
    rest = output(None)
    if rest is None:
        return None
    think = None
    if thinking is not None:
        think = _thinking_format(thinking, message_start, requires_reasoning)
        if think is None:
            return None
    if message_start is None:
        if think is not None:
            rest.insert(0, _optional(think))
        return {"type": "structural_tag", "format": _sequence(*rest)}
    if think is not None and requires_reasoning:
        return {"type": "structural_tag", "format": _sequence(think, *rest)}
    start = output(message_start)
    if start is None:
        return None
    format = _sequence(*start)
    if think is not None:
        format = _one_of([_sequence(think, *rest), format])
    return {"type": "structural_tag", "format": format}

_text_format(schema, excludes)

Constrain raw text, which the excludes end, without changing its parsed value.

Source code in vllm/parser/response_template_grammar.py

def _text_format(schema: Any, excludes: list[str]) -> dict[str, Any] | None:
    """Constrain raw text, which the `excludes` end, without changing its
    parsed value."""
    if schema is True:
        return _ANY_TEXT
    if isinstance(schema, dict) and {"format", "pattern"} & set(schema):
        return _string_regex(schema, excludes)
    if not isinstance(schema, dict) or set(schema) - _ANNOTATIONS - {
        "type",
        "nullable",
        "enum",
        "const",
        "anyOf",
        "oneOf",
    }:
        return None
    branches = [*(schema.get("anyOf") or []), *(schema.get("oneOf") or [])]
    if "enum" in schema or "const" in schema:
        choices = _string_choices(schema)
        if choices is None or branches or set(_schema_types(schema)) - {"string"}:
            return None
        return _const_strings(choices)
    if any(_text_format(branch, excludes) is not _ANY_TEXT for branch in branches):
        return None
    return _ANY_TEXT

_thinking_format(thinking, message_start, nonempty=False)

The reasoning, which holds at least a character if nonempty.

Source code in vllm/parser/response_template_grammar.py

def _thinking_format(
    thinking: ResponseTemplateField,
    message_start: list[str] | None,
    nonempty: bool = False,
) -> dict[str, Any] | None:
    """The reasoning, which holds at least a character if `nonempty`."""
    openers, closers = _openers(thinking, message_start), _closers(thinking)
    if not openers or not closers:
        return None
    text = _sequence(_nonempty_start(closers), _ANY_TEXT) if nonempty else _ANY_TEXT
    return _one_of([_tag(opener, text, closers) for opener in openers])

_typed_by_parser(schema)

Whether the parser reads schema values with their JSON type.

Source code in vllm/parser/response_template_grammar.py

def _typed_by_parser(schema: Any) -> bool:
    """Whether the parser reads schema values with their JSON type."""
    if schema is True or _schema_types(schema):
        return True
    return isinstance(schema, dict) and (
        not set(schema) - _ANNOTATIONS or _string_choices(schema) is not None
    )

_value_format(schema, parameters, json_values, closers)

Use JSON for decoded values and raw text for string values.

Source code in vllm/parser/response_template_grammar.py

def _value_format(
    schema: Any, parameters: dict[str, Any], json_values: bool, closers: list[str]
) -> dict[str, Any] | None:
    """Use JSON for decoded values and raw text for string values."""
    if schema is False:
        return None
    types = _schema_types(schema)
    if json_values or (types and "string" not in types):
        return _json_format(_json_value_schema(schema, parameters), closers)
    return _text_format(schema, closers)

answer_rest_structural_tag(answer)

The rest of an answer that is already open, or None if underivable.

Source code in vllm/parser/response_template_grammar.py

def answer_rest_structural_tag(answer: ResponseTemplateField) -> dict[str, Any] | None:
    """The rest of an `answer` that is already open, or `None` if underivable."""
    closers = _closers(answer)
    if closers is None:
        return None
    return {"type": "structural_tag", "format": _sequence(*_answer_text(closers))}

message_start_anchors(template)

The strings the assistant message starts after, if literal.

Source code in vllm/parser/response_template_grammar.py

def message_start_anchors(template: ResponseTemplate) -> list[str]:
    """The strings the assistant message starts after, if literal."""
    if template.start_anchor_literals:
        return template.start_anchor_literals
    return _literals(template.start_anchor_re.pattern) or []

tool_call_structural_tag(field, functions, *, parallel_tool_calls, thinking=None, answer=None, answer_format=None, message_start=None, requires_reasoning=False, sorted_arguments=False, strict_arguments=False)

Derive native calls to functions in the format of the tool call field, respecting strict arguments, count and reasoning.

functions are OpenAI function definitions. strict_arguments enforces the schema of every function's arguments where the template can. Strict xml-inline arguments come in declared order, or also alphabetical with sorted_arguments. An answer field lets the model answer instead, in answer_format if given; without functions, only the answer is derived. When the model may answer, calls may also name any other function the template can spell, with any arguments, so a harness can tell the model the function does not exist. At the message_start, the tag covers the output from its first token, which opens with non-empty reasoning if requires_reasoning. Returns None if the template format cannot be derived, and raises ValueError for names it cannot encode.

Source code in vllm/parser/response_template_grammar.py

def tool_call_structural_tag(
    field: ResponseTemplateField | None,
    functions: list[dict[str, Any]],
    *,
    parallel_tool_calls: bool,
    thinking: ResponseTemplateField | None = None,
    answer: ResponseTemplateField | None = None,
    answer_format: dict[str, Any] | None = None,
    message_start: list[str] | None = None,
    requires_reasoning: bool = False,
    sorted_arguments: bool = False,
    strict_arguments: bool = False,
) -> dict[str, Any] | None:
    """Derive native calls to `functions` in the format of the tool call
    `field`, respecting strict arguments, count and reasoning.

    `functions` are OpenAI function definitions. `strict_arguments` enforces
    the schema of every function's arguments where the template can. Strict
    `xml-inline` arguments come in declared order, or also alphabetical with
    `sorted_arguments`. An `answer` field lets the model answer instead, in
    `answer_format` if given; without `functions`, only the answer is derived.
    When the model may answer, calls may also name any other function the
    template can spell, with any arguments, so a harness can tell the model
    the function does not exist. At the `message_start`,
    the tag covers the output from its first token, which opens with non-empty
    reasoning if `requires_reasoning`. Returns `None` if the template format
    cannot be derived, and raises `ValueError` for names it cannot encode."""
    if not functions and answer is not None:

        def reply(start: list[str] | None) -> list[dict[str, Any]] | None:
            format = _reply_format(answer, start, answer_format)
            return None if format is None else [format]

        return _structural_tag(reply, thinking, message_start, requires_reasoning)
    if not functions or field is None or not _names_function_by_opener(field):
        return None
    opener = _split_group(field.open_re.pattern, "name")
    closer_literals = _closers(field)
    if opener is None or closer_literals is None:
        return None
    arguments, enforced = [], []
    for function in functions:
        # Like the generic JSON schema, a function without parameters takes none.
        parameters = function.get("parameters") or {"type": "object", "properties": {}}
        strict = bool(function.get("strict"))
        function_arguments = None
        if strict or strict_arguments:
            try:
                function_arguments = _arguments_format(
                    field, parameters, strict=True, sorted_arguments=sorted_arguments
                )
            except ValueError:
                if strict:
                    raise
        enforced.append(function_arguments is not None)
        if function_arguments is None and not strict:
            function_arguments = _arguments_format(field, parameters, strict=False)
        if function_arguments is None:
            return None
        arguments.append(function_arguments)
    names = []
    for function in functions:
        name = function["name"]
        if not re.fullmatch(opener[1], name):
            raise ValueError(f"tool name {name!r} cannot be encoded in tool calls")
        names.append(_escape(name))
    if answer is not None:
        others = _other_names(
            opener[1], [f["name"] for f, e in zip(functions, enforced) if e]
        )
        loose = _arguments_format(field, {}, strict=False)
        if others is not None and loose is not None:
            # Calls to any function whose schema is not enforced take any arguments.
            names = [name for name, e in zip(names, enforced) if e] + [f"(?:{others})"]
            arguments = [args for args, e in zip(arguments, enforced) if e] + [loose]
    closers = _const_strings(closer_literals) if closer_literals else None

    def call(start: list[str] | None) -> dict[str, Any]:
        return _one_of(
            [
                _sequence(
                    {"type": "regex", "pattern": _call_opener(opener, name, start)},
                    function_arguments,
                )
                for name, function_arguments in zip(names, arguments)
            ]
        )

    def output(start: list[str] | None) -> list[dict[str, Any]] | None:
        elements = [call(start)]
        if parallel_tool_calls and field.repeats:
            repeated = _sequence(closers, call(None)) if closers else call(None)
            elements.append({"type": "star", "content": repeated})
        if closers:
            # EOS can terminate a call without writing its closer.
            elements.append(_optional(closers))
        if answer is not None:
            reply = _reply_format(answer, start, answer_format)
            if reply is None:
                return None
            elements = [_one_of([_sequence(*elements), reply])]
        return elements

    return _structural_tag(output, thinking, message_start, requires_reasoning)