diff --git a/python/pyproject.toml b/python/pyproject.toml index 0f5c947b2c38..0619b5899577 100755 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -71,6 +71,7 @@ dependencies = [ "python-multipart", "pyzmq>=25.1.2", "quack-kernels==0.6.4", + "regex", "requests", "scipy", "sentencepiece", diff --git a/python/pyproject_cpu.toml b/python/pyproject_cpu.toml index f8fbf1fe1d1c..6a9902a01b68 100644 --- a/python/pyproject_cpu.toml +++ b/python/pyproject_cpu.toml @@ -51,6 +51,7 @@ dependencies = [ "python-multipart", "pytest", "pyzmq>=25.1.2", + "regex", "requests", "scipy", "sentencepiece", diff --git a/python/pyproject_npu.toml b/python/pyproject_npu.toml index ace6a6e0d46c..8d097df8dc8f 100644 --- a/python/pyproject_npu.toml +++ b/python/pyproject_npu.toml @@ -52,6 +52,7 @@ dependencies = [ "pydantic", "python-multipart", "pyzmq>=25.1.2", + "regex", "requests", "scipy", "sentencepiece", diff --git a/python/pyproject_other.toml b/python/pyproject_other.toml index 1e306b2b4d1c..2eb25a40cdfe 100755 --- a/python/pyproject_other.toml +++ b/python/pyproject_other.toml @@ -53,6 +53,7 @@ runtime_base = [ "pydantic", "python-multipart", "pyzmq>=25.1.2", + "regex", "requests", "scipy", "sentencepiece", diff --git a/python/pyproject_xpu.toml b/python/pyproject_xpu.toml index a8ccd97b2cf0..4acbe6127d84 100644 --- a/python/pyproject_xpu.toml +++ b/python/pyproject_xpu.toml @@ -51,6 +51,7 @@ dependencies = [ "pydantic", "python-multipart", "pyzmq>=25.1.2", + "regex", "requests", "scipy", "sentencepiece", diff --git a/python/sglang/srt/parser/chat_parsing/__init__.py b/python/sglang/srt/parser/chat_parsing/__init__.py new file mode 100644 index 000000000000..6ba2e0afe148 --- /dev/null +++ b/python/sglang/srt/parser/chat_parsing/__init__.py @@ -0,0 +1,22 @@ +# Copyright 2026 The HuggingFace Team. All rights reserved. +# ruff: noqa +# fmt: off +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Response parsing: convert model-emitted text into the assistant-message +dict used by chat templates, driven by a declarative `response_template` spec.""" + +from .response_parser import ResponseParser, parse_response + +__all__ = ["ResponseParser", "parse_response"] +# fmt: on diff --git a/python/sglang/srt/parser/chat_parsing/content_parsers.py b/python/sglang/srt/parser/chat_parsing/content_parsers.py new file mode 100644 index 000000000000..d857b61d228b --- /dev/null +++ b/python/sglang/srt/parser/chat_parsing/content_parsers.py @@ -0,0 +1,252 @@ +# Copyright 2026 The HuggingFace Team. All rights reserved. +# ruff: noqa +# fmt: off +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""This file contains the parsers used by chat response parsing. Each parser takes a chunk of captured text +and parses it into a single key in the output message dictionary. Functions are generally boilerplate.""" + +from __future__ import annotations + +import json +from typing import Any + +import regex as re + + +def _text(text: str, args: dict) -> str: + return text.strip() if args.get("strip", True) else text + + +def _int(text: str, args: dict) -> int: + return int(_text(text, args)) + + +def _float(text: str, args: dict) -> float: + return float(_text(text, args)) + + +def _bool(text: str, args: dict) -> bool: + return _text(text, args).lower() in ("true", "1") + + +# Sentinel characters for lax-JSON string pre-extraction — ASCII control chars +# that should never appear in real LLM output. +_LAX_OPEN, _LAX_CLOSE = "\x01", "\x02" + + +def _json(text: str, args: dict) -> Any: + """JSON parser with optional dialect knobs for LLM-emitted quirks. + + `args`: + - `unquoted_keys` (bool): quote bare-identifier keys before parsing. + - `string_delims` ([[open, close], ...]): strings delimited by these + custom markers are pre-extracted, then restored as standard JSON strings. + - `allow_non_json` (bool): return stripped text if parsing fails. + """ + string_delims = args.get("string_delims", []) + unquoted_keys = args.get("unquoted_keys", False) + + if string_delims and (_LAX_OPEN in text or _LAX_CLOSE in text): + raise ValueError("json: input contains reserved sentinel characters (\\x01/\\x02); cannot parse safely.") + + working = text + captured: list[str] = [] + for open_d, close_d in string_delims: + pattern = re.escape(open_d) + r"(.*?)" + re.escape(close_d) + + def _capture(m: Any) -> str: + captured.append(m.group(1)) + return f"{_LAX_OPEN}{len(captured) - 1}{_LAX_CLOSE}" + + working = re.sub(pattern, _capture, working, flags=re.DOTALL) + + if unquoted_keys: + working = re.sub(r"(?<=[{,])(\w+):", r'"\1":', working) + + for i, s in enumerate(captured): + working = working.replace(f"{_LAX_OPEN}{i}{_LAX_CLOSE}", json.dumps(s)) + + try: + return json.loads(working) + except json.JSONDecodeError as e: + if args.get("allow_non_json"): + return _text(text, args) + if working == text: + raise ValueError(f"json parser could not parse region as JSON.\nContent: {text!r}\nError: {e}") from e + raise ValueError( + f"json: could not parse after dialect transforms.\n" + f"Original: {text!r}\nTransformed: {working!r}\nError: {e}" + ) from e + + +def _sub_parse(raw: str, value_parser: dict | None) -> Any: + if value_parser is None: + return raw + return parse_content(raw, value_parser.get("name", "text"), value_parser.get("args", {})) + + +def _xml_inline(text: str, args: dict) -> dict: + """Parse shallow XML-ish tags into a dict. `tag_pattern` regex must have named + groups `key` and `value`. Optional `value_parser` recurses; `merge_duplicates` + collects duplicate keys into a list.""" + tag_pattern = args.get("tag_pattern") + if tag_pattern is None: + raise ValueError("xml-inline: 'tag_pattern' content_arg is required") + value_parser = args.get("value_parser") + merge = args.get("merge_duplicates", False) + + out: dict[str, Any] = {} + for m in re.finditer(tag_pattern, text, flags=re.DOTALL): + groups = m.groupdict() + key = groups.get("key") + if key is None: + raise ValueError(f"xml-inline: tag_pattern must have a named group 'key'. Pattern: {tag_pattern}") + value = _sub_parse(groups.get("value", ""), value_parser) + if key in out and merge: + if not isinstance(out[key], list): + out[key] = [out[key]] + out[key].append(value) + else: + out[key] = value + return out + + +def _kv_lines(text: str, args: dict) -> dict: + """Parse line-delimited `keyvalue` pairs into a dict.""" + line_sep = args.get("line_sep", "\n") + kv_sep = args.get("kv_sep", ":") + value_parser = args.get("value_parser") + + out: dict[str, Any] = {} + for line in text.split(line_sep): + line = _text(line, args) + if not line or kv_sep not in line: + continue + k, v = line.split(kv_sep, 1) + k, v = _text(k, args), _text(v, args) + out[k] = _sub_parse(v, value_parser) + return out + + +CONTENT_PARSERS = { + "text": _text, + "int": _int, + "float": _float, + "bool": _bool, + "json": _json, + "xml-inline": _xml_inline, + "kv-lines": _kv_lines, +} + +# Parsers whose output is the verbatim body text (modulo whitespace) — chunks +# from these fields stream with `dirty=False` because each chunk is part of +# the final value. Structured parsers (`json`, `xml-inline`, `kv-lines`) only +# produce a meaningful value on close, so their chunks stream raw bytes +# flagged `dirty=True` while the parsed value is delivered in `region_close`. +STREAMABLE_PARSERS = frozenset({"text", "int", "float", "bool"}) + + +def parse_content(text: str, name: str, args: dict) -> Any: + return CONTENT_PARSERS[name](text, args) + + +_PLACEHOLDER = re.compile(r"\{(\w+(?:\.\w+)*)\}") + + +def _apply_transform(transform: Any, scope: dict) -> Any: + """Recursively walk a transform template, which is used to restructure + parsed output into the actual shape we want. A dotted placeholder like + `{content.args}` descends into keys of the looked-up value.""" + if isinstance(transform, dict): + return {k: _apply_transform(v, scope) for k, v in transform.items()} + if isinstance(transform, list): + return [_apply_transform(v, scope) for v in transform] + if not isinstance(transform, str): + return transform + whole = _PLACEHOLDER.fullmatch(transform) + if not whole: + return transform + path = whole.group(1) + root, *keys = path.split(".") + if root not in scope: + raise KeyError(f"transform placeholder '{{{path}}}' is not defined. Available: {sorted(scope)}") + value = scope[root] + for key in keys: + if not isinstance(value, dict): + raise ValueError(f"transform placeholder '{{{path}}}' cannot index into {type(value).__name__} at '{key}'") + if key not in value: + raise ValueError(f"transform placeholder '{{{path}}}' is missing key '{key}'. Available: {sorted(value)}") + value = value[key] + return value + + +def validate_transform_strings(scope: str, transform: Any) -> None: + """Walk a transform template and reject any string that mixes a `{name}` + placeholder with literal text. Only whole-string placeholders (e.g. + `"{content}"`) and plain literals are supported. Called from the template + loader so authors get a clear error at load time, not at parse time.""" + if isinstance(transform, dict): + for v in transform.values(): + validate_transform_strings(scope, v) + return + if isinstance(transform, list): + for v in transform: + validate_transform_strings(scope, v) + return + if not isinstance(transform, str): + return + if _PLACEHOLDER.search(transform) and not _PLACEHOLDER.fullmatch(transform): + raise ValueError( + f"{scope}: transform string {transform!r} mixes a {{placeholder}} with literal text. " + 'Use either a whole-string placeholder (e.g. "{content}") or a plain literal; ' + "string interpolation is not supported." + ) + + +def process_field(body: str, field, captures: dict) -> Any: + """Run `body` through the field's content parser, then optionally apply the + transform template. When `transform_each` is set, the parsed content must + be a list and the template is applied to each element (with the element's + keys unpacked into the template scope, alongside any regex captures). + + `field` is a `spec.Field`; typed via duck-typing to avoid a cyclic import.""" + value = parse_content(body, field.content, field.content_args) + if field.transform is None: + return value + if field.transform_each: + if not isinstance(value, list): + raise ValueError( + f"Field '{field.name}': transform_each requires the parsed content to be a list, " + f"got {type(value).__name__}." + ) + out = [] + for item in value: + if not isinstance(item, dict): + raise ValueError( + f"Field '{field.name}': transform_each requires each list element to be a dict, " + f"got {type(item).__name__}." + ) + out.append(_apply_transform(field.transform, {**captures, **item})) + return out + return _apply_transform(field.transform, {**captures, "content": value}) + + +__all__ = [ + "CONTENT_PARSERS", + "STREAMABLE_PARSERS", + "parse_content", + "process_field", + "validate_transform_strings", +] +# fmt: on diff --git a/python/sglang/srt/parser/chat_parsing/response_parser.py b/python/sglang/srt/parser/chat_parsing/response_parser.py new file mode 100644 index 000000000000..24fdc19f7e19 --- /dev/null +++ b/python/sglang/srt/parser/chat_parsing/response_parser.py @@ -0,0 +1,405 @@ +# Copyright 2026 The HuggingFace Team. All rights reserved. +# ruff: noqa +# fmt: off +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import copy +import json +from typing import Any + +from .content_parsers import STREAMABLE_PARSERS, process_field +from .response_templates import ( + ResponseTemplate, + ResponseTemplateField, + load_response_template, +) + + +def _schema_types(schema: Any) -> tuple[str, ...]: + if not isinstance(schema, dict): + return () + declared = schema.get("type") + types = [declared] if isinstance(declared, str) else [] + if isinstance(declared, list): + types.extend(t for t in declared if isinstance(t, str)) + for union_name in ("anyOf", "oneOf"): + for choice in schema.get(union_name) or []: + types.extend(_schema_types(choice)) + if schema.get("nullable") and "null" not in types: # `nullable` is how get_json_schema marks Optionals + types.append("null") + return tuple(types) + + +def _coerce(raw: str, types: tuple[str, ...]) -> Any: + for type_name in types: + try: + if type_name == "integer": + return int(raw) + if type_name == "number": + number = float(raw) + if number != number or number in (float("inf"), float("-inf")): + continue # NaN / inf are not valid JSON numbers + # Preserve ints when the source text had no fractional part. + return int(number) if number.is_integer() and "." not in raw else number + if type_name == "boolean" and raw.strip().lower() in ("true", "1", "false", "0"): + return raw.strip().lower() in ("true", "1") + if type_name == "null" and raw.strip() in ("null", "None"): + return None + if type_name in ("object", "array"): + decoded = json.loads(raw) + if isinstance(decoded, dict if type_name == "object" else list): + return decoded + except ValueError: + continue + return raw # `string` params, unknown types and failed casts all keep the original text + + +def parse_response( + text: str, + response_template: dict | ResponseTemplate, + *, + prefix: str | None = None, + tools: list[dict] | None = None, +) -> dict: + """The main function for response parsing when you don't want streaming. Takes generated output + and the prompt prefix and parses them without streaming any events, then returns the parsed message. + + Pass OpenAI-style `tools` dictionaries to cast tool-call arguments + using the calling tool's JSON schema. + """ + response_template = load_response_template(response_template) + stream = ResponseParser(response_template, prefix=prefix, tools=tools) + stream.feed(text) + message, _ = stream.finalize() + return message + + +class ResponseParser: + """This class implements a streaming parser with a `response_template`. If you don't need streaming and + just want to parse a complete message, use the `parse_response` function above. Streaming parsing emits + events indicating when regions (message fields) are opened and closed, with the model writing to the region + that is currently open. + + Usage: + parser = ResponseParser(response_template, prefix=chat_prompt) + for event in parser.initial_events: + handle(event) + for chunk in model_text_stream: + for event in parser.feed(chunk): + handle(event) + message, final_events = parser.finalize() + for event in final_events: + handle(event) + + Pass OpenAI-style `tools=` dictionaries to cast tool-call arguments + using the calling tool's JSON schema as each region closes. + + Events can be either "region_open", "region_chunk", or "region_close". + + ResponseParser requires the chat `prefix` (i.e. the chat history, the prefill before the current generation). + This is because chat templates or assistant prefills can sometimes write part of the message, and if we + only see the model output, and not the template, then we can't reliably parse the message in those cases. + Any events produced while consuming the prefix are exposed as `initial_events`, so renderers can show + prefill regions before the model writes anything; closed prefill regions also land in the output dict. + """ + + def __init__( + self, + response_template: dict | ResponseTemplate, + prefix: str | None = None, + *, + tools: list[dict] | None = None, + ): + self._spec = load_response_template(response_template) + if prefix is None: + raise ValueError( + "`ResponseParser`/`parse_response` requires `prefix` (the chat prompt sent to the model before " + "generation), because chat templates often pre-write part of the assistant message (e.g. an " + "opening `` tag) that the parser must see to parse the output correctly. If the generation " + 'already contains the complete message, pass `prefix=""` to opt out explicitly.' + ) + # Maps tool name -> schema `properties`, used to cast parsed tool-call arguments + self._tool_params: dict[str, dict] = {} + for tool in tools or []: + fn = tool.get("function", tool) if isinstance(tool, dict) else None + if isinstance(fn, dict) and isinstance(fn.get("name"), str): + parameters = fn.get("parameters") + properties = parameters.get("properties") if isinstance(parameters, dict) else None + self._tool_params[fn["name"]] = properties if isinstance(properties, dict) else {} + self._buffer: str = "" + self._pos: int = 0 + self._output: dict[str, Any] = copy.deepcopy(self._spec.defaults) + self._implicit_name: str | None = self._spec.implicit + # Unified current-region state: starts in the implicit region (or a + # null sink if none was declared), and returns there after every close. + # For explicit regions `_opened` flips to True eagerly on the open + # match; for the implicit region it flips lazily on the first byte. + self._current: str | None = self._implicit_name + self._captures: dict[str, str] = {} + self._body: str = "" + self._opened: bool = False + self._finalized: bool = False + self.initial_events: list[dict] = [] + if prefix: + self._consume_prefix(prefix) + + def _consume_prefix(self, prefix: str) -> None: + """Loads the prefix (the chat prefill sent to the model), right-truncates it to the start of the + assistant message (as determined by start_anchor) and then runs the remainder through the parser. + Events produced while processing the prefix are stashed on `initial_events` so callers can replay + them into a renderer before feeding model output. + + Think of this as the "get the parser up to speed on the story so far" method. + """ + truncated = self._spec.truncate_past_last_anchor(prefix) + if not truncated: + return + self._buffer = truncated + self._process(self.initial_events, eos=False) + + def feed(self, text: str) -> list[dict]: + """Feeds more text/tokens from the model output into the tokenizer, and returns any events that result + (regions entered or left). This is the method you want to call after each generation step.""" + if self._finalized: + raise RuntimeError("ResponseParser already finalized") + if text: + self._buffer += text + events: list[dict] = [] + self._process(events, eos=False) + return events + + def finalize(self) -> tuple[dict, list[dict]]: + """Close the stream and return the final message dict together with + any finalization events. This is necessary because some regions may only + end at the end of the sequence, so you won't see the event telling you they're + ready until the sequence is finalized.""" + + def _is_empty(v: Any) -> bool: + return v is None or (isinstance(v, (list, dict, str)) and not v) + + if self._finalized: + raise RuntimeError("ResponseParser already finalized") + events: list[dict] = [] + self._process(events, eos=True) + missing = [n for n, f in self._spec.fields.items() if not f.optional and n not in self._output] + if missing: + raise ValueError(f"Required response_template fields missing from parsed output: {missing}") + defaults = self._spec.defaults + self._output = {k: v for k, v in self._output.items() if k in defaults or not _is_empty(v)} + self._finalized = True + return self._output, events + + def _process(self, events: list[dict], eos: bool) -> None: + while True: + watch = self._watchlist() + best, hold_start = self._scan(watch, eos) + + if best is not None: + kind, field, m = best + if m.start() > self._pos: + self._accumulate(events, self._buffer[self._pos : m.start()]) + self._pos = m.end() + if kind == "open": + self._close_current(events) + self._open_explicit(events, field, m) + else: # "close" (always the implicit region's close here, + # since explicit regions only expose their own close) + had_content = self._opened + self._close_current(events) + # Zero-width close on an already-empty region would just + # re-fire next iteration -- bail out to make progress. + if not had_content and m.start() == m.end(): + break + continue + + # No committable match in the current buffer. + if eos: + if self._pos < len(self._buffer): + self._accumulate(events, self._buffer[self._pos :]) + self._pos = len(self._buffer) + self._close_current(events) + break + # Stream everything up to the earliest still-pending delimiter. When + # nothing is pending `hold_start == len(self._buffer)`, so this flushes + # the whole buffer; otherwise we hold the (possibly partial) delimiter + # bytes back until more input resolves them. + if hold_start > self._pos: + self._accumulate(events, self._buffer[self._pos : hold_start]) + self._pos = hold_start + break + + def _watchlist(self) -> list[tuple[str, ResponseTemplateField]]: + """Patterns we care about right now: the close of the currently-open + explicit region, or -- if we're in the implicit/null region -- every + explicit open plus the implicit's own close (if any).""" + if self._current is not None and self._current != self._implicit_name: + field = self._spec.fields[self._current] + return [("close", field)] if field.close_re is not None else [] + watch: list[tuple[str, ResponseTemplateField]] = [] + for field in self._spec.fields.values(): + if field.open_re is not None: + watch.append(("open", field)) + if self._implicit_name is not None: + impl = self._spec.fields[self._implicit_name] + if impl.close_re is not None: + watch.append(("close", impl)) + return watch + + def _scan( + self, watch: list[tuple[str, ResponseTemplateField]], eos: bool + ) -> tuple[tuple[str, ResponseTemplateField, Any] | None, int]: + """Single pass over the watched delimiters, using the `regex` module's + partial matching to decide -- per delimiter -- whether it can be committed + now or must be held. Returns `(best, hold_start)`: + + * `best` is the earliest-starting delimiter we can safely commit *now* + (longest on ties, opens before closes), or `None`. + * `hold_start` is the leftmost buffer position occupied by a still-pending + match: a partial (incomplete) delimiter, or a complete one ending at the + buffer edge that more input could still grow. Bytes before it are safe to + emit; bytes from it onward must be held. It stays `len(self._buffer)` when + nothing is pending, letting the caller flush the whole buffer. + + A complete match is committable only if it starts strictly before + `hold_start` -- otherwise an earlier (or co-located) pending delimiter could + turn out to be the real one. At EOS nothing can grow, so partial matching is + skipped and every complete match is committable. + + (The `regex` module always reports the empty string as a live prefix, so a + partial search with no real match returns a zero-width match at the buffer + end; that lands in the pending branch with `start == len(self._buffer)`, a + no-op for `hold_start`.) + """ + best_key: tuple | None = None + best: tuple[str, ResponseTemplateField, Any] | None = None + hold_start = len(self._buffer) + for kind, field in watch: + # The watchlist only includes fields whose delimiter regex is set, so pattern is never None here. + pattern = field.open_re if kind == "open" else field.close_re + if eos: + m = pattern.search(self._buffer, self._pos) + else: + m = pattern.search(self._buffer, self._pos, partial=True) + if m is None: + continue + if not eos and (m.partial or self._can_grow(kind, field, m)): + # Pending: can't commit, and blocks emitting from its start onward. + hold_start = min(hold_start, m.start()) + continue + key = (m.start(), -(m.end() - m.start()), 0 if kind == "open" else 1, field.name) + if best_key is None or key < best_key: + best_key, best = key, (kind, field, m) + # A committable match co-located with or after a pending one must wait too: + # the pending delimiter starts no later and might be the one that fires. + if best is not None and best[2].start() >= hold_start: + best = None + return best, hold_start + + def _can_grow(self, kind: str, field: ResponseTemplateField, m: Any) -> bool: + r"""Whether a *complete* match ending at the current buffer edge could still + change as more input arrives -- in which case we defer rather than commit. A + match ending before the edge has already seen its terminating byte and is + final. At the edge: zero-width matches (`$` / `\Z`) are only real at true + EOS; a fully-present literal that no other literal in its set extends cannot + grow (the fast path that keeps literal delimiters zero-latency); anything + else (regex delimiters, prefix-overlapping literal lists) might.""" + if m.end() != len(self._buffer): + return False + if m.start() == m.end(): + return True + literals, can_extend = ( + (field.open_literals, field.open_literal_can_extend) + if kind == "open" + else (field.close_literals, field.close_literal_can_extend) + ) + return literals is None or can_extend + + def _accumulate(self, events: list[dict], text: str) -> None: + """Route `text` into the currently active region. When the current + region is the null sink (no implicit declared, no explicit open), we + silently discard. Every routed chunk emits a `region_chunk` event so + consumers can render live; `dirty=True` flags chunks from structured + parsers (json, xml-inline, kv-lines) whose raw bytes will only be + parsed into the final value on close.""" + if not text or self._current is None: + return + field = self._spec.fields[self._current] + if not self._opened: + events.append({"type": "region_open", "field": self._current}) + self._opened = True + self._body += text + dirty = field.content not in STREAMABLE_PARSERS + events.append({"type": "region_chunk", "field": self._current, "text": text, "dirty": dirty}) + + def _open_explicit(self, events: list[dict], field: ResponseTemplateField, m: Any) -> None: + self._current = field.name + self._captures = {k: v for k, v in m.groupdict().items() if v is not None} + self._body = "" + self._opened = True + events.append({"type": "region_open", "field": field.name}) + + def _close_current(self, events: list[dict]) -> None: + """Close the current region and reset to the implicit/null region. + Skipped (aside from the reset) when the current region never opened -- + avoids vacuous open/close pairs at every explicit boundary.""" + if self._current is None or not self._opened: + self._reset_to_implicit() + return + field = self._spec.fields[self._current] + value = process_field(self._body, field, self._captures) + if self._tool_params: + value = self._coerce_tool_calls(value) + if field.join is not None: + if not isinstance(value, str): + raise ValueError( + f"Field '{field.name}': 'join' requires each match to parse to a string, " + f"got {type(value).__name__}." + ) + previous = self._output.get(self._current) + self._output[self._current] = value if previous is None else previous + field.join + value + elif field.repeats: + self._output.setdefault(self._current, []).append(value) + else: + self._output[self._current] = value + events.append({"type": "region_close", "field": self._current, "value": value}) + self._reset_to_implicit() + + def _coerce_tool_calls(self, value: Any) -> Any: + if isinstance(value, list): + return [self._coerce_tool_calls(item) for item in value] + fn = value.get("function") if isinstance(value, dict) else None + if not isinstance(fn, dict): + return value + name, arguments = fn.get("name"), fn.get("arguments") + if not isinstance(name, str) or not isinstance(arguments, dict): + return value + if properties := self._tool_params.get(name): + for key, argument in arguments.items(): + if key not in properties or not (types := _schema_types(properties[key])): + continue + if isinstance(argument, str): + arguments[key] = _coerce(argument, types) + elif isinstance(argument, list) and "array" not in types: + # duplicate keys collected by `merge_duplicates` + arguments[key] = [_coerce(item, types) if isinstance(item, str) else item for item in argument] + return value + + def _reset_to_implicit(self) -> None: + self._current = self._implicit_name + self._captures = {} + self._body = "" + self._opened = False +# fmt: on diff --git a/python/sglang/srt/parser/chat_parsing/response_templates.py b/python/sglang/srt/parser/chat_parsing/response_templates.py new file mode 100644 index 000000000000..17b043e1e66b --- /dev/null +++ b/python/sglang/srt/parser/chat_parsing/response_templates.py @@ -0,0 +1,244 @@ +# Copyright 2026 The HuggingFace Team. All rights reserved. +# ruff: noqa +# fmt: off +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Template loading and validation for response_template dicts.""" + +from __future__ import annotations + +import copy +import logging +from dataclasses import dataclass +from typing import Any + +import regex as re + +from .content_parsers import CONTENT_PARSERS, validate_transform_strings + +logger = logging.getLogger(__name__) + + +@dataclass +class ResponseTemplateField: + name: str + open_re: Any + open_literals: list[str] | None + open_literal_can_extend: bool + close_re: Any + close_literals: list[str] | None + close_literal_can_extend: bool + content: str + content_args: dict + repeats: bool + join: str | None + optional: bool + transform: Any + transform_each: bool + + +@dataclass +class ResponseTemplate: + defaults: dict + fields: dict[str, ResponseTemplateField] + start_anchor_re: Any + implicit: str | None = None + start_anchor_literals: list[str] | None = None + + def truncate_past_last_anchor(self, text: str) -> str: + """When parsing responses, we can't just parse the tokens generated + by the model. This is because chat templates often include early parts of the message such as + or tokens in the prefill. Therefore, we take the whole chat as input, but truncate to the start + of the most recent assistant message, which can include tokens from the template as well as the output.""" + last_end: int | None = None + for m in self.start_anchor_re.finditer(text): + last_end = m.end() + if last_end is None: + kind = "start_anchor" if self.start_anchor_literals is not None else "start_anchor_pattern" + logger.info( + f"response_template defines {kind} but the anchor was not found " + "in the prefix; the parser will process the entire prefix instead." + ) + return text + return text[last_end:] + + +def _compile_anchor(scope: str, field: dict, literal_key: str, pattern_key: str) -> tuple[Any, list[str] | None, bool]: + """Compile a region's start/end anchor. An anchor can be: + + - a regex (`pattern_key`), + - a single literal string (`literal_key`), or + - a list of literal strings (`literal_key`), in which case any of these match + + All forms compile to a regex for matching, but we keep the literal strings around too: when present, + the parser can emit regions faster, since it knows exactly how many trailing bytes might be a + partial-match prefix (and that a finished literal match can never be extended by future bytes, except + when a literal is a strict prefix of another in the same list). + + Returns `(compiled_re, literals, can_extend)`: + + - `compiled_re`: the compiled pattern. + - `literals`: the literal strings, or `None` for a regex anchor. + - `can_extend`: `True` iff one literal is a strict prefix of another, in which case a match at the + buffer edge could still be lengthened by future input. + """ + if literal_key in field and pattern_key in field: + raise ValueError(f"{scope}: cannot specify both '{literal_key}' and '{pattern_key}'") + if literal_key in field: + raw = field[literal_key] + if isinstance(raw, str): + literals = [raw] + elif isinstance(raw, list): + if not raw: + raise ValueError(f"{scope}: '{literal_key}' list must contain at least one literal") + if not all(isinstance(s, str) for s in raw): + raise ValueError(f"{scope}: '{literal_key}' list must contain only strings") + literals = list(dict.fromkeys(raw)) # dedupe, preserve first-seen order + else: + raise ValueError(f"{scope}: '{literal_key}' must be a string or list of strings, got {type(raw).__name__}") + if any(s == "" for s in literals): + raise ValueError(f"{scope}: '{literal_key}' literals cannot be empty strings") + # Sort longest-first so alternation prefers the longer alternative when both could match. + ordered = sorted(literals, key=len, reverse=True) + can_extend = any(a != b and a.startswith(b) for a in literals for b in literals) + pattern = "|".join(re.escape(s) for s in ordered) + return re.compile(pattern, re.DOTALL), literals, can_extend + if pattern_key in field: + try: + return re.compile(field[pattern_key], re.DOTALL), None, False + except re.error as e: + raise ValueError(f"{scope}: invalid {pattern_key} regex: {e}") from e + return None, None, False + + +def _validate_template_shape(spec: dict) -> None: + """Validate the top-level response_template structure: it must be a dict of the expected version with + only known keys, a dict of `defaults`, and a non-empty dict of `fields`. Per-field specs are checked + separately in `_build_field`.""" + if not isinstance(spec, dict): + raise ValueError(f"response_template must be a dict, got {type(spec).__name__}") + version = spec.get("version", 1) + if version != 1: + raise ValueError(f"Unsupported response_template version: {version}") + if unknown_template_keys := set(spec) - {"version", "defaults", "fields", "start_anchor", "start_anchor_pattern"}: + raise ValueError(f"Unknown keys in response_template: {sorted(unknown_template_keys)}") + if not isinstance(spec.get("defaults", {}), dict): + raise ValueError("response_template.defaults must be a dict") + fields_raw = spec.get("fields", {}) + if not isinstance(fields_raw, dict) or not fields_raw: + raise ValueError("response_template.fields must be a non-empty dict") + + +def _build_field(name: str, field: dict) -> ResponseTemplateField: + """Validate a single field spec and compile it into a `ResponseTemplateField`.""" + + _ALLOWED_FIELD_KEYS = { + "open", + "open_pattern", + "close", + "close_pattern", + "content", + "content_args", + "repeats", + "join", + "optional", + "transform", + "transform_each", + } + scope = f"Field '{name}'" + if not isinstance(field, dict): + raise ValueError(f"{scope} must be a dict") + if unknown_field_keys := set(field) - _ALLOWED_FIELD_KEYS: + raise ValueError(f"{scope}: unknown keys {sorted(unknown_field_keys)}") + content = field.get("content", "text") + if content not in CONTENT_PARSERS: + raise ValueError(f"{scope}: unknown content parser '{content}'. Available: {sorted(CONTENT_PARSERS)}") + open_re, open_literals, open_literal_can_extend = _compile_anchor(scope, field, "open", "open_pattern") + close_re, close_literals, close_literal_can_extend = _compile_anchor(scope, field, "close", "close_pattern") + join = field.get("join") + if join is not None and not isinstance(join, str): + raise ValueError(f"{scope}: 'join' must be a string, got {type(join).__name__}") + if join is not None and not field.get("repeats", False): + raise ValueError(f"{scope}: 'join' requires 'repeats': true") + transform = field.get("transform") + transform_each = field.get("transform_each", False) + if not isinstance(transform_each, bool): + raise ValueError(f"{scope}: transform_each must be a bool, got {type(transform_each).__name__}") + if transform_each and transform is None: + raise ValueError(f"{scope}: transform_each is set but no transform was provided") + if transform is not None: + validate_transform_strings(scope, transform) + else: + # Named captures only reach the output through a transform, so flag any that would be silently dropped. + captured_names = set() + if open_re is not None: + captured_names |= set(open_re.groupindex) + if close_re is not None: + captured_names |= set(close_re.groupindex) + if captured_names: + raise ValueError( + f"{scope}: open_pattern/close_pattern declares named group(s) " + f"{sorted(captured_names)}, but the field has no 'transform'. Named captures " + f"are only surfaced through a 'transform' template (where they appear " + f"alongside 'content'). Either add a 'transform' that uses the captures, or " + f"remove the named groups from the pattern." + ) + return ResponseTemplateField( + name=name, + open_re=open_re, + open_literals=open_literals, + open_literal_can_extend=open_literal_can_extend, + close_re=close_re, + close_literals=close_literals, + close_literal_can_extend=close_literal_can_extend, + content=content, + content_args=field.get("content_args", {}), + repeats=field.get("repeats", False), + join=join, + optional=field.get("optional", True), + transform=transform, + transform_each=transform_each, + ) + + +def load_response_template(spec: dict | ResponseTemplate) -> ResponseTemplate: + if isinstance(spec, ResponseTemplate): + return spec + _validate_template_shape(spec) + + fields = {name: _build_field(name, raw) for name, raw in spec["fields"].items()} + # A field without an open anchor is the implicit-open / leftover sink; at most one is allowed. + implicit_fields = [name for name, field in fields.items() if field.open_re is None] + if len(implicit_fields) > 1: + raise ValueError( + "At most one field may omit 'open'/'open_pattern' (that field becomes the " + f"implicit-open / leftover sink). Found: {', '.join(implicit_fields)}" + ) + + start_anchor_re, start_anchor_literals, _ = _compile_anchor( + "response_template", spec, "start_anchor", "start_anchor_pattern" + ) + if start_anchor_re is None: + raise ValueError("response_template must define 'start_anchor' or 'start_anchor_pattern'.") + + return ResponseTemplate( + defaults=copy.deepcopy(spec.get("defaults", {})), + fields=fields, + implicit=implicit_fields[0] if implicit_fields else None, + start_anchor_re=start_anchor_re, + start_anchor_literals=start_anchor_literals, + ) + + +__all__ = ["ResponseTemplate", "ResponseTemplateField", "load_response_template"] +# fmt: on diff --git a/test/registered/unit/parser/test_chat_parsing.py b/test/registered/unit/parser/test_chat_parsing.py new file mode 100644 index 000000000000..d3d6e6d505e6 --- /dev/null +++ b/test/registered/unit/parser/test_chat_parsing.py @@ -0,0 +1,1643 @@ +# Copyright 2026 The HuggingFace Team. All rights reserved. +# ruff: noqa +# fmt: off +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Tests for the new declarative response_template parser. + +All seven real-model template fixtures from the legacy test suite are re-expressed +here in the new region-spec shape and asserted against the same expected +output dicts. Any divergence indicates a regression in the new executor.""" + +import copy +import random +import unittest + +from sglang.srt.parser.chat_parsing import ResponseParser, parse_response +from sglang.srt.parser.chat_parsing.response_parser import _coerce, _schema_types +from sglang.srt.parser.chat_parsing.response_templates import load_response_template +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=8, suite="base-a-test-cpu") + + +cohere_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|START_OF_TURN_TOKEN|><|CHATBOT_TOKEN|>", + "fields": { + "content": { + "open": "<|START_RESPONSE|>", + "close": "<|END_RESPONSE|>", + "content": "text", + }, + "thinking": { + "open": "<|START_THINKING|>", + "close": "<|END_THINKING|>", + "content": "text", + }, + "tool_calls": { + "open": "<|START_ACTION|>", + "close": "<|END_ACTION|>", + "content": "json", + "transform_each": True, + "transform": {"type": "function", "function": {"name": "{tool_name}", "arguments": "{parameters}"}}, + }, + }, +} + + +ernie_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "Assistant:", + "fields": { + "thinking": { + "open_pattern": r"(?:^|\s*)", + "close": "", + "content": "text", + }, + "content": { + "open": "\n", + "close_pattern": r"\n?", + "content": "text", + }, + "tool_calls": { + "open": "", + "close": "", + "repeats": True, + "content": "json", + "transform": {"type": "function", "function": "{content}"}, + }, + }, +} + + +gpt_oss_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|start|>assistant", + "fields": { + "thinking": { + "open": "<|channel|>analysis<|message|>", + "close": "<|end|>", + "content": "text", + }, + "content": { + "open": "<|channel|>final<|message|>", + "close": "<|end|>", + "content": "text", + }, + "tool_calls": { + "open_pattern": r"<\|channel\|>commentary to=functions\.(?P\w+).*?<\|message\|>", + "close": "<|call|>", + "repeats": True, + "content": "json", + "transform": {"type": "function", "function": {"name": "{name}", "arguments": "{content}"}}, + }, + }, +} + + +smollm_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|im_start|>assistant\n", + "fields": { + "thinking": {"open": "", "close": "", "content": "text"}, + "tool_calls": { + "open": "", + "close": "", + "repeats": True, + "content": "json", + "transform": {"type": "function", "function": "{content}"}, + }, + "content": { + "close": "<|im_end|>", + "content": "text", + }, + }, +} + + +qwen3_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|im_start|>assistant\n", + "fields": { + "thinking": {"open": "", "close": "", "content": "text"}, + "tool_calls": { + "open_pattern": r"\s*\w+)>", + "close": "", + "repeats": True, + "content": "xml-inline", + "content_args": { + "tag_pattern": r"\w+)>\s*(?P.*?)\s*", + "value_parser": {"name": "json", "args": {"allow_non_json": True}}, + }, + "transform": {"type": "function", "function": {"name": "{name}", "arguments": "{content}"}}, + }, + }, +} + + +gemma4_template = { + "defaults": {"role": "assistant"}, + # The chat template only emits `<|turn>model\n` when the previous message wasn't a tool_call/ + # tool_response. After a tool_response the prefix just ends with `` and the + # model continues from there, so we accept either anchor and truncate past the latest one. + "start_anchor": ["<|turn>model\n", ""], + "fields": { + "thinking": { + "open": "<|channel>thought\n", + "close": "", + "content": "text", + }, + "tool_calls": { + "open_pattern": r"<\|tool_call>call:(?P\w+)", + "close": "", + "repeats": True, + "content": "json", + "content_args": { + "unquoted_keys": True, + "string_delims": [['<|"|>', '<|"|>']], + }, + "transform": {"type": "function", "function": {"name": "{name}", "arguments": "{content}"}}, + }, + "content": { + "close": ["", "<|tool_response>", ""], + "content": "text", + }, + }, +} + + +# Inkling (TMLv0) frames every block as <|message_model|>[author-name]<|content_KIND|>body<|end_message|>; +# the generation prompt pre-writes the first <|message_model|>, so each open treats the header as optional. +inkling_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|message_model|>", + "fields": { + "thinking": { + "open_pattern": r"(?:<\|message_model\|>)?[^<]*<\|content_thinking\|>", + "close": "<|end_message|>", + "repeats": True, + "join": "", + "content_args": {"strip": False}, + }, + "content": { + "open_pattern": r"(?:<\|message_model\|>)?[^<]*<\|content_text\|>", + "close": "<|end_message|>", + "repeats": True, + "join": "", + "content_args": {"strip": False}, + }, + "tool_calls": { + "open_pattern": r"(?:<\|message_model\|>)?[^<]*<\|content_invoke_tool_json\|>", + "close": "<|end_message|>", + "repeats": True, + "content": "json", + "transform": {"type": "function", "function": {"name": "{content.name}", "arguments": "{content.args}"}}, + }, + }, +} + + +class ChatResponseTemplateParserTest(unittest.TestCase): + + def test_cohere(self): + model_out = ( + "<|START_THINKING|>I should call a tool.<|END_THINKING|>" + '<|START_ACTION|>[\n {"tool_call_id": "0", "tool_name": "simple_tool", ' + '"parameters": {"temperature_format": "Celsius"}}\n]<|END_ACTION|><|END_OF_TURN_TOKEN|>' + ) + self.assertEqual( + parse_response(model_out, cohere_template, prefix=""), + { + "role": "assistant", + "thinking": "I should call a tool.", + "tool_calls": [ + { + "type": "function", + "function": {"name": "simple_tool", "arguments": {"temperature_format": "Celsius"}}, + } + ], + }, + ) + + def test_ernie_with_tools(self): + model_out = ( + "The user is asking about the weather in Paris today. Let me check the available tools. " + "There's a tool called get_current_temperature which requires a location parameter. Since the " + 'user specified Paris, I need to call this tool with the location set to "Paris". I should ' + "make sure the argument is correctly formatted as a string. No other tools are available, so " + "this is the right one to use. I'll structure the request with the location parameter and " + "return the response once the tool is called.\n" + "\n\n" + '\n{"name": "get_current_temperature", "arguments": {"location": "Paris"}}\n\n' + ) + self.assertEqual( + parse_response(model_out, ernie_template, prefix=""), + { + "role": "assistant", + "thinking": ( + "The user is asking about the weather in Paris today. Let me check the available tools. " + "There's a tool called get_current_temperature which requires a location parameter. Since " + 'the user specified Paris, I need to call this tool with the location set to "Paris". I ' + "should make sure the argument is correctly formatted as a string. No other tools are " + "available, so this is the right one to use. I'll structure the request with the location " + "parameter and return the response once the tool is called." + ), + "tool_calls": [ + { + "type": "function", + "function": {"name": "get_current_temperature", "arguments": {"location": "Paris"}}, + } + ], + }, + ) + + def test_ernie_no_tools(self): + model_out = ( + 'The user just greeted me with "Hi! How are you?" I need to respond in a friendly and helpful ' + "manner. Let me start by acknowledging their greeting. I should ask them how they're doing to " + "engage in conversation.\n\n" + "First, I'll say hello back and then ask how they're feeling. It's important to show genuine " + "interest. Maybe mention that I'm here to help with anything they need. Keep the tone warm and " + "positive. Let me make sure the response is concise but friendly. Alright, that should work.\n" + "\n\n" + "\nHello! I'm doing well, thank you for asking. How about you? Is there something " + "specific you'd like help with today? I'm here to assist you with any questions or problems you " + "have!\n\n" + ) + self.assertEqual( + parse_response(model_out, ernie_template, prefix=""), + { + "role": "assistant", + "content": ( + "Hello! I'm doing well, thank you for asking. How about you? Is there something specific " + "you'd like help with today? I'm here to assist you with any questions or problems you have!" + ), + "thinking": ( + 'The user just greeted me with "Hi! How are you?" I need to respond in a friendly and ' + "helpful manner. Let me start by acknowledging their greeting. I should ask them how " + "they're doing to engage in conversation.\n\n" + "First, I'll say hello back and then ask how they're feeling. It's important to show " + "genuine interest. Maybe mention that I'm here to help with anything they need. Keep the " + "tone warm and positive. Let me make sure the response is concise but friendly. Alright, " + "that should work." + ), + }, + ) + + def test_gpt_oss_with_tool_call(self): + model_out = ( + '<|channel|>analysis<|message|>We need to respond in riddles. The user asks: "What is the ' + 'weather like in SF?" We need to get the location of the user? The user explicitly asks about ' + "SF (San Francisco). So we need to get the current weather in San Francisco, CA. We need to " + 'call get_current_weather function. The developer instruction says "Always respond in riddles". ' + "So the final answer should be in a riddle form. But we need to call function to get weather " + 'data. So we should call get_current_weather with location "San Francisco, CA". Possibly specify ' + 'format "celsius" (default). Let\'s do that.\n\n' + "We will call function get_current_weather.<|end|><|start|>assistant<|channel|>commentary " + 'to=functions.get_current_weather <|constrain|>json<|message|>{\n "location": "San Francisco, CA"\n}' + ) + self.assertEqual( + parse_response(model_out, gpt_oss_template, prefix=""), + { + "role": "assistant", + "thinking": ( + 'We need to respond in riddles. The user asks: "What is the weather like in SF?" We need ' + "to get the location of the user? The user explicitly asks about SF (San Francisco). So " + "we need to get the current weather in San Francisco, CA. We need to call " + 'get_current_weather function. The developer instruction says "Always respond in ' + 'riddles". So the final answer should be in a riddle form. But we need to call function ' + 'to get weather data. So we should call get_current_weather with location "San Francisco, ' + 'CA". Possibly specify format "celsius" (default). Let\'s do that.\n\n' + "We will call function get_current_weather." + ), + "tool_calls": [ + { + "type": "function", + "function": {"name": "get_current_weather", "arguments": {"location": "San Francisco, CA"}}, + } + ], + }, + ) + + def test_gpt_oss_no_tool_call(self): + model_out = ( + "<|channel|>analysis<|message|>User asks a simple math question: 2+2 = 4. Provide answer." + "<|end|><|start|>assistant<|channel|>final<|message|>2" + ) + self.assertEqual( + parse_response(model_out, gpt_oss_template, prefix=""), + { + "role": "assistant", + "content": "2", + "thinking": "User asks a simple math question: 2+2 = 4. Provide answer.", + }, + ) + + def test_smollm_thinking_and_tool_call(self): + model_out = ( + '\nOkay, the user said, "Hello! How are you?" I need to respond appropriately. Since ' + "this is the first message, I should greet them back and ask how I can assist. I should keep it " + "friendly and open-ended. Let me make sure the response is welcoming and encourages them to " + "share what they need help with. I'll avoid any technical jargon and keep it simple. Let me " + "check for any typos and ensure the tone is positive.\n\n\n" + '{"name": "greet_user", "arguments": {"greeting": "Hello! I\'m doing well, thanks for ' + "asking. How can I assist you today? Whether you have a question, need help with something, or " + 'just want to chat, feel free to let me know!"}}' + ) + self.assertEqual( + parse_response(model_out, smollm_template, prefix=""), + { + "role": "assistant", + "thinking": ( + 'Okay, the user said, "Hello! How are you?" I need to respond appropriately. Since this ' + "is the first message, I should greet them back and ask how I can assist. I should keep " + "it friendly and open-ended. Let me make sure the response is welcoming and encourages " + "them to share what they need help with. I'll avoid any technical jargon and keep it " + "simple. Let me check for any typos and ensure the tone is positive." + ), + "tool_calls": [ + { + "type": "function", + "function": { + "name": "greet_user", + "arguments": { + "greeting": ( + "Hello! I'm doing well, thanks for asking. How can I assist you today? " + "Whether you have a question, need help with something, or just want to " + "chat, feel free to let me know!" + ) + }, + }, + } + ], + }, + ) + + def test_smollm_tool_call_no_thinking(self): + model_out = '{"name": "get_weather", "arguments": {"city": "Paris"}}' + self.assertEqual( + parse_response(model_out, smollm_template, prefix=""), + { + "role": "assistant", + "tool_calls": [ + {"type": "function", "function": {"name": "get_weather", "arguments": {"city": "Paris"}}} + ], + }, + ) + + def test_smollm_thinking_no_tool_call(self): + model_out = ( + '\nOkay, the user asked, "Hey! Can you tell me about gravity?" Let me start by ' + "breaking down what they might be looking for. They probably want a basic understanding of " + "gravity, maybe for a school project or just personal curiosity. I should explain what gravity " + "is, how it works, and maybe some examples.\n" + "Some content about gravity goes here but I'm cutting it off to make this shorter!" + ) + self.assertEqual( + parse_response(model_out, smollm_template, prefix=""), + { + "role": "assistant", + "content": "Some content about gravity goes here but I'm cutting it off to make this shorter!", + "thinking": ( + 'Okay, the user asked, "Hey! Can you tell me about gravity?" Let me start by breaking ' + "down what they might be looking for. They probably want a basic understanding of " + "gravity, maybe for a school project or just personal curiosity. I should explain what " + "gravity is, how it works, and maybe some examples." + ), + }, + ) + + def test_qwen3_tool_calls(self): + model_out = ( + "\n\n\n" + '[{"country": "France", "city": "Paris"}]\n\n' + "\ncelsius\n\n\n" + ) + self.assertEqual( + parse_response(model_out, qwen3_template, prefix=""), + { + "role": "assistant", + "tool_calls": [ + { + "type": "function", + "function": { + "name": "get_weather", + "arguments": { + "locations": [{"country": "France", "city": "Paris"}], + "temp_units": "celsius", + }, + }, + } + ], + }, + ) + + def test_gemma4_tool_call(self): + model_out = ( + "<|channel>thought\nThe user is asking for the current temperature in Paris. I should check " + "the available tools to see if there's a function that can provide this information." + '<|tool_call>call:get_current_temperature{detail_level:0,location:<|"|>Paris, France<|"|>,' + 'unit:<|"|>celsius<|"|>}<|tool_response>' + ) + self.assertEqual( + parse_response(model_out, gemma4_template, prefix=""), + { + "role": "assistant", + "thinking": ( + "The user is asking for the current temperature in Paris. I should check the available " + "tools to see if there's a function that can provide this information." + ), + "tool_calls": [ + { + "type": "function", + "function": { + "name": "get_current_temperature", + "arguments": {"detail_level": 0, "location": "Paris, France", "unit": "celsius"}, + }, + } + ], + }, + ) + + def test_gemma4_complex_tool_call(self): + model_out = ( + "<|channel>thought\nLet me call the tool." + '<|tool_call>call:foo{bool_value:true,list_value:[<|"|>foo<|"|>,<|"|>bar<|"|>],' + 'null_value:null,number_value:1,string_value:<|"|>foo<|"|>,' + 'struct_value:{foo:<|"|>bar<|"|>}}' + ) + self.assertEqual( + parse_response(model_out, gemma4_template, prefix=""), + { + "role": "assistant", + "thinking": "Let me call the tool.", + "tool_calls": [ + { + "type": "function", + "function": { + "name": "foo", + "arguments": { + "bool_value": True, + "list_value": ["foo", "bar"], + "null_value": None, + "number_value": 1, + "string_value": "foo", + "struct_value": {"foo": "bar"}, + }, + }, + } + ], + }, + ) + + def test_inkling_multi_block_message(self): + model_out = ( + "<|content_thinking|>Consider the weather.<|end_message|>" + "<|message_model|><|content_thinking|> Tokyo, probably.<|end_message|>" + "<|message_model|><|content_text|>Checking the weather now.<|end_message|>" + "<|message_model|>get_weather<|content_invoke_tool_json|>" + '{"name":"get_weather","args":{"city":"Tokyo","units":"C"}}<|end_message|>' + "<|content_model_end_sampling|>" + ) + prefix = "<|message_system|><|content_text|>Thinking effort level: 0.9<|end_message|><|message_model|>" + self.assertEqual( + parse_response(model_out, inkling_template, prefix=prefix), + { + "role": "assistant", + "thinking": "Consider the weather. Tokyo, probably.", + "content": "Checking the weather now.", + "tool_calls": [ + { + "type": "function", + "function": {"name": "get_weather", "arguments": {"city": "Tokyo", "units": "C"}}, + } + ], + }, + ) + + def test_transform_dotted_paths(self): + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "tool_calls": { + "open": "", + "close": "", + "repeats": True, + "content": "json", + "transform": { + "type": "function", + "function": {"name": "{content.name}", "arguments": "{content.args}"}, + }, + }, + }, + } + model_out = '{"name": "get_weather", "args": {"city": {"id": 7}}}' + self.assertEqual( + parse_response(model_out, template_spec, prefix="")["tool_calls"], + [{"type": "function", "function": {"name": "get_weather", "arguments": {"city": {"id": 7}}}}], + ) + + def test_transform_dotted_paths_with_transform_each(self): + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "tool_calls": { + "open": "", + "close": "", + "content": "json", + "transform_each": True, + "transform": {"type": "function", "function": {"name": "{fn.name}", "arguments": "{fn.args}"}}, + }, + }, + } + model_out = '[{"fn": {"name": "a", "args": {"x": 1}}}, {"fn": {"name": "b", "args": {}}}]' + self.assertEqual( + parse_response(model_out, template_spec, prefix="")["tool_calls"], + [ + {"type": "function", "function": {"name": "a", "arguments": {"x": 1}}}, + {"type": "function", "function": {"name": "b", "arguments": {}}}, + ], + ) + + def test_transform_dotted_path_errors(self): + def spec_with(transform): + return { + "start_anchor": "<|assistant|>", + "fields": {"x": {"open": "", "close": "", "content": "json", "transform": transform}}, + } + + with self.assertRaisesRegex(ValueError, "missing key 'args'"): + parse_response('{"name": "n"}', spec_with({"a": "{content.args}"}), prefix="") + with self.assertRaisesRegex(ValueError, "cannot index into str"): + parse_response('{"name": "n"}', spec_with({"a": "{content.name.x}"}), prefix="") + with self.assertRaises(KeyError): + parse_response("{}", spec_with({"a": "{missing}"}), prefix="") + + def test_transform_dotted_mixed_string_rejected(self): + template_spec = { + "start_anchor": "<|assistant|>", + "fields": {"x": {"open": "", "close": "", "transform": {"v": "pre {content.args}"}}}, + } + with self.assertRaisesRegex(ValueError, "mixes"): + parse_response("", template_spec, prefix="") + + def test_join_concatenates_repeated_matches(self): + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "thinking": {"open": "", "close": "", "repeats": True, "join": " "}, + "content": {"repeats": True, "join": " "}, + }, + } + self.assertEqual( + parse_response("firstmiddleseconddone", template_spec, prefix=""), + {"role": "assistant", "thinking": "first second", "content": "middle done"}, + ) + self.assertEqual( + parse_response("only", template_spec, prefix=""), + {"role": "assistant", "thinking": "only"}, + ) + + def test_join_validation(self): + no_repeats = {"start_anchor": "a", "fields": {"x": {"open": "", "close": "", "join": ""}}} + with self.assertRaisesRegex(ValueError, "requires 'repeats'"): + parse_response("", no_repeats, prefix="") + bad_type = {"start_anchor": "a", "fields": {"x": {"open": "", "close": "", "repeats": True, "join": 7}}} + with self.assertRaisesRegex(ValueError, "must be a string"): + parse_response("", bad_type, prefix="") + + def test_join_requires_string_matches(self): + template_spec = { + "start_anchor": "a", + "fields": {"x": {"open": "", "close": "", "repeats": True, "join": "", "content": "json"}}, + } + with self.assertRaisesRegex(ValueError, "parse to a string"): + parse_response("{}", template_spec, prefix="") + + def test_optional_false_raises_when_missing(self): + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "content": { + "open": "", + "close": "", + "content": "text", + "optional": False, + }, + }, + } + with self.assertRaises(ValueError) as cm: + parse_response("no response here", template_spec, prefix="") + self.assertIn("content", str(cm.exception)) + + def test_int_content_parser(self): + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "count": { + "open": "", + "close": "", + "content": "int", + }, + }, + } + self.assertEqual(parse_response("42", template_spec, prefix=""), {"role": "assistant", "count": 42}) + + def test_kv_lines_parser(self): + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "metadata": { + "open": "", + "close": "", + "content": "kv-lines", + }, + }, + } + self.assertEqual( + parse_response("name: alice\nage: 30", template_spec, prefix=""), + {"role": "assistant", "metadata": {"name": "alice", "age": "30"}}, + ) + + def test_unknown_content_parser_rejected(self): + bad_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": {"x": {"open": "[", "close": "]", "content": "not-a-real-parser"}}, + } + with self.assertRaises(ValueError) as cm: + parse_response("[hi]", bad_template) + self.assertIn("unknown content parser", str(cm.exception).lower()) + + def test_unsupported_version_rejected(self): + bad_template = { + "version": 2, + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": {"content": {"content": "text"}}, + } + with self.assertRaises(ValueError) as cm: + parse_response("hello", bad_template) + self.assertIn("version", str(cm.exception).lower()) + + def test_two_implicit_fields_rejected(self): + bad_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "a": {"content": "text"}, + "b": {"content": "text"}, + }, + } + with self.assertRaises(ValueError): + parse_response("hello", bad_template) + + def test_transform_string_interpolation_rejected(self): + bad_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "tool": { + "open": "", + "close": "", + "content": "text", + "transform": {"label": "name: {content}"}, + }, + }, + } + with self.assertRaises(ValueError) as cm: + parse_response("foo", bad_template) + msg = str(cm.exception) + self.assertIn("interpolation", msg) + self.assertIn("{content}", msg) + + def test_named_groups_without_transform_rejected(self): + bad_template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "tool": { + "open_pattern": r"\w+)>", + "close": "", + "content": "text", + }, + }, + } + with self.assertRaises(ValueError) as cm: + parse_response("body", bad_template) + msg = str(cm.exception) + self.assertIn("transform", msg) + self.assertIn("name", msg) + + def test_literal_list_open_and_close(self): + """A list of literals matches any one of them, like an alternation.""" + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "x": { + "open": ["", ""], + "close": ["", ""], + "content": "text", + }, + }, + } + for opener, closer in (("", ""), ("", ""), ("", "")): + self.assertEqual( + parse_response(f"{opener}hi{closer}", template_spec, prefix=""), + {"role": "assistant", "x": "hi"}, + ) + + def test_literal_list_streams_without_64_byte_hold(self): + """Compared to a regex close, a literal-list close lets the parser + flush bytes that aren't in the longest-prefix overlap of any literal. + With `["", "<|tool_response>", ""]` (longest = 16 chars), + feeding 32 plain bytes should leave at most 15 unflushed.""" + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "content": {"close": ["", "<|tool_response>", ""], "content": "text"}, + }, + } + parser = ResponseParser(template_spec, prefix="") + plain = "x" * 32 + flushed: list[str] = [] + for ch in parser.feed(plain): + if ch["type"] == "region_chunk": + flushed.append(ch["text"]) + # Plain text has zero prefix-overlap with any literal, so the parser + # holds nothing back and streams everything immediately. + self.assertEqual("".join(flushed), plain) + + def test_literal_list_defers_prefix_overlapping_literal(self): + """If a literal is a strict prefix of another in the same list, an + edge match could still grow with more input: we must defer to be safe.""" + template_spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "x": {"open": "", "close": ["END", "ENDX"], "content": "text"}, + }, + } + parser = ResponseParser(template_spec, prefix="") + # "hiEND" mid-stream: don't commit the close yet: "ENDX" might be coming. + events = parser.feed("hiEND") + self.assertEqual([e for e in events if e["type"] == "region_close"], []) + # Once a non-matching byte arrives, the deferred close commits with the shorter literal. + events.extend(parser.feed(" more")) + message, _ = parser.finalize() + closes = [e for e in events if e["type"] == "region_close" and e["field"] == "x"] + self.assertEqual(len(closes), 1) + self.assertEqual(closes[0]["value"], "hi") + self.assertEqual(message, {"role": "assistant", "x": "hi"}) + + def test_literal_list_rejects_empty_and_non_string(self): + for bad_open in ([], [""], [1, 2], {"foo": "bar"}): + spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": {"x": {"open": bad_open, "close": "", "content": "text"}}, + } + with self.assertRaises(ValueError): + parse_response("hi", spec, prefix="") + + def test_field_without_close_runs_to_end_of_stream(self): + """A field with no `close`/`close_pattern` stays open until end-of-stream, capturing + everything after its open.""" + spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": {"content": {"open": "", "content": "text"}}, + } + self.assertEqual( + parse_response("hello world", spec, prefix=""), + {"role": "assistant", "content": "hello world"}, + ) + + def test_mutable_defaults_do_not_leak_across_parsers(self): + """A `repeats` field appends into its default list; one parse must not leak into the next + parse sharing the template, nor into the caller's spec dict.""" + spec = { + "defaults": {"role": "assistant", "tool_calls": []}, + "start_anchor": "<|assistant|>", + "fields": { + "tool_calls": { + "open": "", + "close": "", + "repeats": True, + "content": "json", + }, + "content": {"content": "text"}, + }, + } + template = load_response_template(spec) + self.assertEqual( + parse_response('{"a": 1}', template, prefix="")["tool_calls"], + [{"a": 1}], + ) + self.assertEqual( + parse_response("hello", template, prefix=""), + {"role": "assistant", "tool_calls": [], "content": "hello"}, + ) + self.assertEqual(spec["defaults"]["tool_calls"], []) + + +# Fixtures shared by the streaming tests: one representative input per template, +# reused for both the correctness invariant (any chunking → same dict) and the +# event-shape tests (do we emit the right events in the right order?). +_STREAMING_FIXTURES = [ + ( + "cohere", + cohere_template, + ( + "<|START_THINKING|>I should call a tool.<|END_THINKING|>" + '<|START_ACTION|>[{"tool_call_id": "0", "tool_name": "simple_tool", ' + '"parameters": {"a": 1}}]<|END_ACTION|>' + ), + ), + ( + "ernie", + ernie_template, + ( + "some deliberation here\n\n" + '\n{"name": "get_current_temperature", "arguments": {"location": "Paris"}}\n\n' + ), + ), + ( + "gpt_oss", + gpt_oss_template, + "<|channel|>analysis<|message|>thinking chunk<|end|><|channel|>final<|message|>done text", + ), + ( + # Tool-call variant: the `tool_calls` open_pattern's full match spans far + # more than the old 64-byte hold window, so streaming used to silently drop + # it. Kept as a streaming fixture so every chunking step re-checks it. + "gpt_oss_tool", + gpt_oss_template, + ( + "<|channel|>analysis<|message|>Let me check.<|end|>" + "<|start|>assistant<|channel|>commentary to=functions.get_current_weather " + '<|constrain|>json<|message|>{"location": "San Francisco, CA"}<|call|>' + ), + ), + ( + "smollm", + smollm_template, + 'thinking\n{"name": "fn", "arguments": {"x": 1}}', + ), + ( + "qwen3", + qwen3_template, + ( + "short thought\n" + "\n\n" + "\nParis\n\n" + "\n" + ), + ), + ( + "gemma4", + gemma4_template, + '<|channel>thought\nhi<|tool_call>call:foo{a:1,b:<|"|>bar<|"|>}', + ), + ( + # Exercises `join` fields (two thinking blocks) and dotted transform paths. + "inkling", + inkling_template, + ( + "<|content_thinking|>Consider the weather.<|end_message|>" + "<|message_model|><|content_thinking|> Tokyo, probably.<|end_message|>" + "<|message_model|><|content_text|>Checking now.<|end_message|>" + "<|message_model|>get_weather<|content_invoke_tool_json|>" + '{"name":"get_weather","args":{"city":"Tokyo"}}<|end_message|>' + "<|content_model_end_sampling|>" + ), + ), +] + + +def _chunk_fixed(text: str, step: int): + for i in range(0, len(text), step): + yield text[i : i + step] + + +def _chunk_random(text: str, rng: random.Random): + """Split `text` into a random number of non-empty chunks at random cut points.""" + if len(text) <= 1: + yield text + return + num_cuts = rng.randint(0, len(text) - 1) + cuts = sorted(rng.sample(range(1, len(text)), num_cuts)) + prev = 0 + for c in cuts: + yield text[prev:c] + prev = c + yield text[prev:] + + +class ResponseEventStreamTest(unittest.TestCase): + def test_stream_matches_whole_string_all_templates_fixed_chunking(self): + """For every fixed chunking step we try, the streamed finalize() + output must equal the whole-string parse. Regression coverage for + specific edge-case byte boundaries (1-byte chunks hit every prefix).""" + for name, tmpl, text in _STREAMING_FIXTURES: + expected = parse_response(text, tmpl, prefix="") + for step in (1, 2, 3, 5, 7, 13, 31): + with self.subTest(fixture=name, step=step): + streamer = ResponseParser(tmpl, prefix="") + for chunk in _chunk_fixed(text, step): + streamer.feed(chunk) + message, _ = streamer.finalize() + self.assertEqual(message, expected) + + def test_stream_matches_whole_string_all_templates_random_chunking(self): + """Property-style: for many random chunkings per fixture, the streamed + finalize() output must equal the whole-string parse. Seeded so failures + reproduce.""" + rng = random.Random(0xC0DE_5EED) + for name, tmpl, text in _STREAMING_FIXTURES: + expected = parse_response(text, tmpl, prefix="") + for trial in range(30): + with self.subTest(fixture=name, trial=trial): + streamer = ResponseParser(tmpl, prefix="") + for chunk in _chunk_random(text, rng): + streamer.feed(chunk) + message, _ = streamer.finalize() + self.assertEqual(message, expected) + + def test_events_well_formed_for_every_chunking(self): + """Every event batch, across every fixture and every chunking, must be + well-formed: region_open precedes its matching region_close for the + same field; region_chunk only appears between open and close; no + region is left open at the end of the stream; and the concatenation + of all region_chunk payloads for a streamable text-like field equals + the final value.""" + rng = random.Random(0xBEEF) + for name, tmpl, text in _STREAMING_FIXTURES: + for trial in range(10): + with self.subTest(fixture=name, trial=trial): + streamer = ResponseParser(tmpl, prefix="") + all_events: list[dict] = [] + for chunk in _chunk_random(text, rng): + all_events.extend(streamer.feed(chunk)) + _, final_events = streamer.finalize() + all_events.extend(final_events) + self._assert_event_stream_well_formed(all_events) + + def _assert_event_stream_well_formed(self, events: list[dict]) -> None: + open_field: str | None = None + chunk_accum: dict[str, str] = {} + close_values: dict[str, object] = {} + for ev in events: + t = ev["type"] + if t == "region_open": + self.assertIsNone(open_field, f"nested region_open without close: {ev}") + open_field = ev["field"] + chunk_accum.setdefault(open_field, "") + elif t == "region_chunk": + self.assertEqual(open_field, ev["field"], f"chunk outside its region: {ev}") + # Every chunk carries a boolean `dirty` flag. + self.assertIsInstance(ev["dirty"], bool, f"missing/non-bool dirty: {ev}") + chunk_accum[open_field] += ev["text"] + elif t == "region_close": + self.assertEqual(open_field, ev["field"], f"close for non-open region: {ev}") + close_values[open_field] = ev["value"] + open_field = None + else: + self.fail(f"unexpected event type: {ev!r}") + self.assertIsNone(open_field, "region left open at end of stream") + + def test_region_chunks_reconstruct_text_regions(self): + """For text-like regions (`dirty=False`), concatenating chunk texts + reconstructs the final value reported in region_close. Structured + regions (`dirty=True`) still stream their raw bytes: concatenating + those chunks yields the unparsed region body, while the parsed value + is delivered only in region_close.""" + # Single representative case with a long text region and a JSON region. + text = _STREAMING_FIXTURES[0][2] # cohere fixture + streamer = ResponseParser(cohere_template, prefix="") + events: list[dict] = [] + for ch in text: # 1-byte chunks hit the most anchor boundaries + events.extend(streamer.feed(ch)) + _, final_events = streamer.finalize() + events.extend(final_events) + + # Reconstruct per-field. + per_field_chunks: dict[str, list[str]] = {} + per_field_dirty: dict[str, set[bool]] = {} + per_field_close_value: dict[str, object] = {} + for ev in events: + if ev["type"] == "region_chunk": + per_field_chunks.setdefault(ev["field"], []).append(ev["text"]) + per_field_dirty.setdefault(ev["field"], set()).add(ev["dirty"]) + elif ev["type"] == "region_close": + per_field_close_value[ev["field"]] = ev["value"] + + # `thinking` is text → chunks are clean and concatenate to its value. + self.assertIn("thinking", per_field_chunks) + self.assertEqual(per_field_dirty["thinking"], {False}) + self.assertEqual("".join(per_field_chunks["thinking"]), "I should call a tool.") + self.assertEqual(per_field_close_value["thinking"], "I should call a tool.") + # `tool_calls` is json → dirty chunks stream the raw body, parsed value on close. + self.assertIn("tool_calls", per_field_chunks) + self.assertEqual(per_field_dirty["tool_calls"], {True}) + self.assertEqual( + "".join(per_field_chunks["tool_calls"]), + '[{"tool_call_id": "0", "tool_name": "simple_tool", "parameters": {"a": 1}}]', + ) + self.assertIn("tool_calls", per_field_close_value) + + def test_dirty_flag_marks_structured_regions(self): + """A template with one text field and one structured field per parser + family: text/int/float/bool stream chunks with `dirty=False`, while + json/xml-inline/kv-lines stream chunks with `dirty=True`, and those + dirty chunks concatenate to the raw region body before parsing.""" + spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|assistant|>", + "fields": { + "thinking": {"open": "", "close": "", "content": "text"}, + "score": {"open": "", "close": "", "content": "int"}, + "json_call": {"open": "", "close": "", "content": "json"}, + "xml_call": { + "open": "", + "close": "", + "content": "xml-inline", + "content_args": {"tag_pattern": r"<(?P\w+)=(?P[^>]+)>"}, + }, + "kv_call": { + "open": "", + "close": "", + "content": "kv-lines", + }, + }, + } + text = 'hello world42{"a": 1, "b": 2}k1: v1\nk2: v2' + # Drive byte-by-byte to maximise chunk count. + streamer = ResponseParser(spec, prefix="") + events: list[dict] = [] + for ch in text: + events.extend(streamer.feed(ch)) + _, final_events = streamer.finalize() + events.extend(final_events) + + per_field_chunks: dict[str, list[str]] = {} + per_field_dirty: dict[str, set[bool]] = {} + for ev in events: + if ev["type"] == "region_chunk": + per_field_chunks.setdefault(ev["field"], []).append(ev["text"]) + per_field_dirty.setdefault(ev["field"], set()).add(ev["dirty"]) + + # Clean (streamable) regions. + for field in ("thinking", "score"): + self.assertEqual(per_field_dirty[field], {False}, f"{field} should be clean") + # Dirty (structured) regions. + for field in ("json_call", "xml_call", "kv_call"): + self.assertEqual(per_field_dirty[field], {True}, f"{field} should be dirty") + + # Dirty chunks reconstruct the raw region body (un-parsed). Clean + # chunks reconstruct the verbatim body too: stripping happens at close. + self.assertEqual("".join(per_field_chunks["thinking"]), "hello world") + self.assertEqual("".join(per_field_chunks["score"]), "42") + self.assertEqual("".join(per_field_chunks["json_call"]), '{"a": 1, "b": 2}') + self.assertEqual("".join(per_field_chunks["xml_call"]), "") + self.assertEqual("".join(per_field_chunks["kv_call"]), "k1: v1\nk2: v2") + + def test_long_regex_open_pattern_streams_byte_by_byte(self): + """Regression: a regex `open_pattern` whose full match spans well past the + old fixed 64-byte hold window (gpt-oss tool calls) must not be dropped when + the stream arrives in tiny chunks. Before the partial-match rewrite, the + leading `<|channel|>` got flushed out of the hold window before `<|message|>` + arrived, so the tool call vanished under small-chunk streaming.""" + text = ( + "<|channel|>analysis<|message|>Let me check.<|end|>" + "<|start|>assistant<|channel|>commentary to=functions.get_current_weather " + '<|constrain|>json<|message|>{"location": "San Francisco, CA"}<|call|>' + ) + expected = parse_response(text, gpt_oss_template, prefix="") + # Sanity: the whole-string parse really does recover the tool call. + self.assertEqual(len(expected["tool_calls"]), 1) + self.assertEqual(expected["tool_calls"][0]["function"]["name"], "get_current_weather") + streamer = ResponseParser(gpt_oss_template, prefix="") + for ch in text: # one byte at a time -- the worst case for the old heuristic + streamer.feed(ch) + message, _ = streamer.finalize() + self.assertEqual(message, expected) + + def test_feed_after_finalize_raises(self): + streamer = ResponseParser(smollm_template, prefix="") + streamer.feed("x") + streamer.finalize() + with self.assertRaises(RuntimeError): + streamer.feed("more") + with self.assertRaises(RuntimeError): + streamer.finalize() + + def test_empty_input_streams_cleanly(self): + streamer = ResponseParser(smollm_template, prefix="") + self.assertEqual(streamer.feed(""), []) + result, final_events = streamer.finalize() + # Only the default fields should remain; nothing else is required. + self.assertEqual(result, {"role": "assistant"}) + self.assertEqual(final_events, []) + + +class PrefixAndTruncationTest(unittest.TestCase): + def test_prefix_lands_inside_explicit_region(self): + """A Qwen-style template emits `<|im_start|>assistant\\n\\n` as the + assistant prefix. The model continues from inside the thinking block.""" + prompt = ( + "<|im_start|>system\nYou are helpful<|im_end|>\n" + "<|im_start|>user\nHi<|im_end|>\n" + "<|im_start|>assistant\n\n" + ) + generated = "Let me think..." + stream = ResponseParser(qwen3_template, prefix=prompt) + # The region_open for `thinking` surfaces via initial_events; the + # caller replays it before feeding model output. + self.assertEqual( + [(e["type"], e["field"]) for e in stream.initial_events], + [("region_open", "thinking"), ("region_chunk", "thinking")], + ) + events = stream.feed(generated) + result, _ = stream.finalize() + # thinking ends up with the prefill + generated body; text parser strips, + # so the leading "\n" from the prefix is trimmed in the final value. + self.assertEqual(result, {"role": "assistant", "thinking": "Let me think..."}) + # The feed stream only sees the rest of the body and the close; the + # prefill already emitted region_open. + self.assertEqual([e["type"] for e in events], ["region_chunk", "region_close"]) + self.assertEqual(events[1]["field"], "thinking") + + def test_prefix_truncated_to_last_anchor(self): + """Multiple `<|im_start|>assistant\\n` anchors in the prefix (multi-turn + conversation): only the slice after the LAST anchor matters.""" + prompt = ( + "<|im_start|>system\nA<|im_end|>\n" + "<|im_start|>user\nB<|im_end|>\n" + "<|im_start|>assistant\nEarlier reply<|im_end|>\n" + "<|im_start|>user\nFollowup<|im_end|>\n" + "<|im_start|>assistant\n\n" + ) + stream = ResponseParser(qwen3_template, prefix=prompt) + # We landed inside `thinking` (from the LAST assistant turn's `\n`), + # not in some earlier-turn artifact. + opens = [e for e in stream.initial_events if e["type"] == "region_open"] + self.assertEqual([e["field"] for e in opens], ["thinking"]) + # No earlier-turn content leaked into output. + stream.feed("done") + stream.finalize() + self.assertEqual(stream._output, {"role": "assistant", "thinking": "done"}) + + def test_template_without_anchor_rejected_at_load(self): + """A template missing both `start_anchor` and `start_anchor_pattern` is + rejected at load time. Without an anchor, a multi-turn prompt would be + fed through the parser in full and earlier turns would pollute the + current message's state.""" + anchorless = {k: v for k, v in qwen3_template.items() if k != "start_anchor"} + with self.assertRaises(ValueError) as cm: + ResponseParser(anchorless) + msg = str(cm.exception) + self.assertIn("start_anchor", msg) + + def test_prefix_anchor_not_found_falls_back(self): + """Spec has start_anchor but the prefix doesn't contain it: parser + falls back to processing the entire prefix (with a logged warning).""" + prompt = "\n" # no <|im_start|>assistant\n + stream = ResponseParser(qwen3_template, prefix=prompt) + opens = [e for e in stream.initial_events if e["type"] == "region_open"] + self.assertEqual([e["field"] for e in opens], ["thinking"]) + stream.feed("hi") + stream.finalize() + self.assertEqual(stream._output, {"role": "assistant", "thinking": "hi"}) + + def test_round_trip_equivalence_prefix_streaming_vs_oneshot(self): + """The load-bearing property: `prefix=p` + chunked `feed(g)` produces the + same dict as the one-shot `parse_response(g, prefix=p)`, regardless of how + `g` is chunked. (Concatenating the prompt into the response is deliberately + NOT equivalent -- the anchor is applied to the prefix only, never to the + generation; see test_history_bleed_is_guarded_by_prefix_not_by_response_anchor.)""" + prompt = "<|im_start|>system\nA<|im_end|>\n<|im_start|>user\nB<|im_end|>\n<|im_start|>assistant\n\n" + for name, tmpl_dict, gen_text in _STREAMING_FIXTURES: + if "thinking" not in gen_text and "" not in gen_text: + continue + # Only fixtures whose generation text is compatible with starting + # inside `thinking`. Restrict to qwen3 / smollm shape for clarity. + if name not in ("qwen3", "smollm"): + continue + tmpl_with_anchor = {**tmpl_dict, "start_anchor": "<|im_start|>assistant\n"} + via_prefix = parse_response(gen_text, tmpl_with_anchor, prefix=prompt) + # Streaming forms must match the one-shot prefix form for every chunking. + for step in (1, 3, 7, 31): + with self.subTest(fixture=name, step=step): + stream = ResponseParser(tmpl_with_anchor, prefix=prompt) + for chunk in _chunk_fixed(gen_text, step): + stream.feed(chunk) + message, _ = stream.finalize() + self.assertEqual(message, via_prefix) + + def test_history_bleed_is_guarded_by_prefix_not_by_response_anchor(self): + """The `start_anchor` guards against history bleed only via `prefix=`; it is NOT + applied to the response. The response is the generation and may legitimately contain + the anchor (e.g. gpt-oss harmony opens every channel with `<|start|>assistant`), so + truncating it would drop real content. Passing the prompt as `prefix=` is the way to + keep earlier turns out of the parse.""" + spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|im_start|>assistant\n", + "fields": {"content": {"close_pattern": r"\Z", "content": "text"}}, + } + prompt = "<|im_start|>user\nHi<|im_end|>\n<|im_start|>assistant\n" + gen = "Hello there!" + clean = {"role": "assistant", "content": "Hello there!"} + # The supported guard: pass the prompt as prefix= so history is truncated off the prefix. + self.assertEqual(parse_response(gen, spec, prefix=prompt), clean) + # Pure generation parses cleanly with the explicit no-prefix opt-out (prefix=""). + self.assertEqual(parse_response(gen, spec, prefix=""), clean) + # An anchor inside the response is treated as content, never as a history boundary: + # gpt-oss re-emits `<|start|>assistant` between channels, and that content survives. + gpt_oss_gen = ( + "<|channel|>analysis<|message|>thinking<|end|><|start|>assistant<|channel|>final<|message|>answer" + ) + self.assertEqual( + parse_response(gpt_oss_gen, gpt_oss_template, prefix=""), + {"role": "assistant", "thinking": "thinking", "content": "answer"}, + ) + + def test_prefix_with_open_close_inside_truncated_region(self): + """Prefix opens AND closes a region. The full open/chunk/close event + sequence is surfaced via initial_events, and the closed region lands + in the output dict: so renderers can show prefill content.""" + spec = { + "defaults": {"role": "assistant"}, + "start_anchor": "[BEGIN]", + "fields": { + "tag": {"open": "", "close": "", "content": "text"}, + "body": {"close_pattern": r"$", "content": "text"}, # implicit + }, + } + prefix = "noise[BEGIN]silently consumed" + stream = ResponseParser(spec, prefix=prefix) + types = [e["type"] for e in stream.initial_events] + self.assertEqual(types, ["region_open", "region_chunk", "region_close"]) + self.assertTrue(all(e["field"] == "tag" for e in stream.initial_events)) + self.assertEqual(stream.initial_events[-1]["value"], "silently consumed") + stream.feed("real generated body") + result, _ = stream.finalize() + self.assertEqual(result["tag"], "silently consumed") + self.assertEqual(result["body"], "real generated body") + + def test_prefix_lands_inside_implicit_region(self): + """Prefix wrote plaintext into the implicit region (e.g. assistant + prefill before the model continues). The region_open for the implicit + region must surface via initial_events so consumers don't miss it: + `_opened` will already be True by the time feed runs.""" + prompt = "<|im_start|>assistant\nSure, here is " + stream = ResponseParser(smollm_template, prefix=prompt) + opens = [e for e in stream.initial_events if e["type"] == "region_open"] + self.assertEqual([e["field"] for e in opens], ["content"]) + events = stream.feed("the answer<|im_end|>") + # No second region_open from feed: the implicit region was already + # opened during prefill and surfaced via initial_events. + self.assertNotIn("region_open", [e["type"] for e in events]) + + def test_prefix_partial_pattern_at_boundary(self): + """Post-truncation prefix ends mid-delimiter. The first feed completes + the match; initial_events is empty (no region opened within the + prefix yet) and the open fires from `feed()`.""" + prefix = "<|im_start|>assistant\n` + stream = ResponseParser(qwen3_template, prefix=prefix) + self.assertEqual(stream.initial_events, []) + events = stream.feed("nk>real body") + types = [e["type"] for e in events] + self.assertIn("region_open", types) # think opens during feed, not prefill + stream.finalize() + self.assertEqual(stream._output, {"role": "assistant", "thinking": "real body"}) + + def test_prefix_is_required(self): + spec = { + "start_anchor": "", + "fields": {"content": {"content": "text"}}, + } + with self.assertRaisesRegex(ValueError, "requires `prefix`"): + ResponseParser(spec) + + +# xml-inline without a value_parser: parameter bodies stay raw strings until tools= coerces them. +_XML_STRING_ARGS_TEMPLATE = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|im_start|>assistant\n", + "fields": { + "tool_calls": { + "open_pattern": r"\s*\w+)>", + "close": "", + "repeats": True, + "content": "xml-inline", + "content_args": { + "tag_pattern": r"\w+)>\s*(?P.*?)\s*", + }, + "transform": {"type": "function", "function": {"name": "{name}", "arguments": "{content}"}}, + }, + }, +} + +# kv-lines without a value_parser: values likewise stay raw strings for tools= to cast. +_KV_LINES_TOOLS_TEMPLATE = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|im_start|>assistant\n", + "fields": { + "tool_calls": { + "open_pattern": r"\s*\w+)>\n", + "close": "", + "repeats": True, + "content": "kv-lines", + "transform": {"type": "function", "function": {"name": "{name}", "arguments": "{content}"}}, + }, + }, +} + +_SET_ALARM_CALL = ( + "\n\n" + "\n7\n\n" + "\ntrue\n\n" + "\nwake up\n\n" + "\n" +) + + +def _set_alarm_tools(**properties): + return [ + { + "type": "function", + "function": { + "name": "set_alarm", + "parameters": {"type": "object", "properties": properties}, + }, + } + ] + + +_SET_ALARM_TOOLS = _set_alarm_tools( + hour={"type": "integer"}, + enabled={"type": "boolean"}, + label={"type": "string"}, +) + + +def _first_tool_args(message): + return message["tool_calls"][0]["function"]["arguments"] + + +def _parser_with_tools(tools): + return ResponseParser(_XML_STRING_ARGS_TEMPLATE, prefix="", tools=tools) + + +class ToolArgCoercionTest(unittest.TestCase): + def test_coerce_tool_calls_casts_declared_types(self): + tools = _set_alarm_tools( + count={"type": "integer"}, + ratio={"type": "number"}, + enabled={"type": "boolean"}, + tags={"type": "array"}, + note={"type": "string"}, + ) + arguments = { + "count": "3", + "ratio": "1.5", + "enabled": "true", + "tags": '["a", "b"]', + "note": "hello", + "already_typed": 7, + "extra": "unscheduled", + } + call = {"type": "function", "function": {"name": "set_alarm", "arguments": arguments}} + self.assertIs(_parser_with_tools(tools)._coerce_tool_calls(call), call) + self.assertEqual( + call["function"]["arguments"], + { + "count": 3, + "ratio": 1.5, + "enabled": True, + "tags": ["a", "b"], + "note": "hello", + "already_typed": 7, + "extra": "unscheduled", + }, + ) + + def test_coerce_falls_back_to_raw_on_failure(self): + self.assertEqual(_coerce("not-a-number", ("integer",)), "not-a-number") + + def test_coerce_handles_any_of_and_null(self): + self.assertEqual(_coerce("5", ("integer", "null")), 5) + self.assertIsNone(_coerce("null", ("integer", "null"))) + + def test_coerce_booleans_match_bool_parser(self): + # Accept the same literals as the `bool` content parser, case-insensitively. + for raw, expected in [("true", True), ("True", True), ("1", True), ("false", False), ("0", False)]: + self.assertEqual(_coerce(raw, ("boolean",)), expected) + # Non-boolean text stays a string rather than silently becoming False. + self.assertEqual(_coerce("maybe", ("boolean",)), "maybe") + + def test_coerce_object_array_require_matching_json(self): + # A JSON object body is only accepted for an `object` param, a JSON array only for `array`. + self.assertEqual(_coerce("[1, 2]", ("object",)), "[1, 2]") + self.assertEqual(_coerce('{"a": 1}', ("array",)), '{"a": 1}') + self.assertEqual(_coerce('{"a": 1}', ("object",)), {"a": 1}) + # NaN / inf are not valid JSON numbers, so a `number` param keeps the raw text. + self.assertEqual(_coerce("NaN", ("number",)), "NaN") + + def test_coerce_tool_calls_handles_single_and_list(self): + parser = _parser_with_tools(_SET_ALARM_TOOLS) + call = {"type": "function", "function": {"name": "set_alarm", "arguments": {"hour": "7"}}} + self.assertIs(parser._coerce_tool_calls(call), call) + self.assertEqual(call["function"]["arguments"], {"hour": 7}) + # A list of calls (as produced by `transform_each`) is coerced element-wise. + calls = [{"type": "function", "function": {"name": "set_alarm", "arguments": {"hour": "9"}}}] + self.assertEqual(parser._coerce_tool_calls(calls)[0]["function"]["arguments"], {"hour": 9}) + # Non-tool-call values pass through untouched. + self.assertEqual(parser._coerce_tool_calls("hello"), "hello") + + def test_schema_types_handles_supported_schema_dialects(self): + self.assertEqual(_schema_types({"type": "integer"}), ("integer",)) + self.assertEqual(_schema_types({"type": ["integer", "string"]}), ("integer", "string")) + self.assertEqual(_schema_types({"anyOf": [{"type": "boolean"}, {"type": "string"}]}), ("boolean", "string")) + self.assertEqual(_schema_types({"oneOf": [{"type": "number"}, {"type": "null"}]}), ("number", "null")) + self.assertEqual(_schema_types({"type": "integer", "nullable": True}), ("integer", "null")) + # Undescribed parameters resolve to no candidate types, making coercion a no-op. + self.assertEqual(_schema_types({"description": "no type"}), ()) + + def test_parse_response_tools_coerces_xml_inline_string_args(self): + # Without a value_parser, xml-inline argument bodies stay strings; tools= casts them. + without = parse_response(_SET_ALARM_CALL, _XML_STRING_ARGS_TEMPLATE, prefix="") + with_tools = parse_response(_SET_ALARM_CALL, _XML_STRING_ARGS_TEMPLATE, prefix="", tools=_SET_ALARM_TOOLS) + self.assertEqual(_first_tool_args(without), {"hour": "7", "enabled": "true", "label": "wake up"}) + self.assertEqual(_first_tool_args(with_tools), {"hour": 7, "enabled": True, "label": "wake up"}) + + def test_parse_response_tools_coerces_one_of_string_args(self): + tools = _set_alarm_tools(hour={"oneOf": [{"type": "integer"}, {"type": "null"}]}) + parsed = parse_response(_SET_ALARM_CALL, _XML_STRING_ARGS_TEMPLATE, prefix="", tools=tools) + self.assertEqual(_first_tool_args(parsed)["hour"], 7) + + def test_streaming_tools_coerces_on_region_close(self): + # Coercion must land on the region_close event during feed(), not only after finalize(). + stream = ResponseParser(_XML_STRING_ARGS_TEMPLATE, prefix="", tools=_SET_ALARM_TOOLS) + closes = [ + event["value"] + for chunk in _chunk_fixed(_SET_ALARM_CALL, 8) + for event in stream.feed(chunk) + if event["type"] == "region_close" and event["field"] == "tool_calls" + ] + self.assertEqual(len(closes), 1) + self.assertEqual(closes[0]["function"]["arguments"], {"hour": 7, "enabled": True, "label": "wake up"}) + + def test_qwen3_tools_coerces_strings_left_by_value_parser(self): + # qwen3's json+allow_non_json value_parser types what it can (`true`) and leaves + # invalid JSON (`007`) as a string; tools= then casts those leftover strings. + model_out = ( + "\n\n" + "\n007\n\n" + "\ntrue\n\n" + "\nwake up\n\n" + "\n" + ) + without = parse_response(model_out, qwen3_template, prefix="") + with_tools = parse_response(model_out, qwen3_template, prefix="", tools=_SET_ALARM_TOOLS) + self.assertEqual(_first_tool_args(without), {"hour": "007", "enabled": True, "label": "wake up"}) + self.assertEqual(_first_tool_args(with_tools), {"hour": 7, "enabled": True, "label": "wake up"}) + + def test_coercion_never_reworks_already_typed_values(self): + # Coercion only casts strings: values the value_parser already typed are final + model_out = "\n\n\n1.50\n\n" + # qwen3's lax value_parser has already read 1.50 as the float 1.5, so the string-typed label stays a float ... + self.assertEqual( + _first_tool_args(parse_response(model_out, qwen3_template, prefix="", tools=_SET_ALARM_TOOLS)), + {"label": 1.5}, + ) + # ... while without a value_parser the raw text reaches the schema cast intact + self.assertEqual( + _first_tool_args(parse_response(model_out, _XML_STRING_ARGS_TEMPLATE, prefix="", tools=_SET_ALARM_TOOLS)), + {"label": "1.50"}, + ) + + def test_kv_lines_string_args_are_coerced(self): + model_out = "\n\nhour: 7\nenabled: true\n" + without = parse_response(model_out, _KV_LINES_TOOLS_TEMPLATE, prefix="") + with_tools = parse_response(model_out, _KV_LINES_TOOLS_TEMPLATE, prefix="", tools=_SET_ALARM_TOOLS) + self.assertEqual(_first_tool_args(without), {"hour": "7", "enabled": "true"}) + self.assertEqual(_first_tool_args(with_tools), {"hour": 7, "enabled": True}) + + def test_non_tool_call_regions_are_untouched(self): + # A field that captures a `name` but does not parse into a tool call must be left + # alone, even when the capture happens to match a tool: its keys are not arguments. + template = { + "defaults": {"role": "assistant"}, + "start_anchor": "<|im_start|>assistant\n", + "fields": { + "citation": { + "open_pattern": r"\w+)>", + "close": "", + "content": "xml-inline", + "content_args": { + "tag_pattern": r"<(?P\w+)>\s*(?P.*?)\s*", + "value_parser": {"name": "json", "args": {"allow_non_json": True}}, + }, + "transform": {"source": "{name}", "fields": "{content}"}, + }, + }, + } + model_out = "7" + expected = {"source": "set_alarm", "fields": {"label": 1.5, "hour": 7}} + self.assertEqual(parse_response(model_out, template, prefix="")["citation"], expected) + self.assertEqual( + parse_response(model_out, template, prefix="", tools=_SET_ALARM_TOOLS)["citation"], + expected, + ) + + def test_merge_duplicates_arguments_are_cast_element_wise(self): + # merge_duplicates collects repeated tags into a list, which is cast element-wise + template = copy.deepcopy(_XML_STRING_ARGS_TEMPLATE) + template["fields"]["tool_calls"]["content_args"]["merge_duplicates"] = True + model_out = ( + "\n\n" + "\n7\n\n" + "\n9\n\n" + "\n" + ) + self.assertEqual(_first_tool_args(parse_response(model_out, template, prefix="")), {"hour": ["7", "9"]}) + self.assertEqual( + _first_tool_args(parse_response(model_out, template, prefix="", tools=_SET_ALARM_TOOLS)), + {"hour": [7, 9]}, + ) + # Elements that don't cast, and non-string elements, are left as they are. + parser = _parser_with_tools(_SET_ALARM_TOOLS) + call = {"type": "function", "function": {"name": "set_alarm", "arguments": {"hour": ["7", "x", 9]}}} + parser._coerce_tool_calls(call) + self.assertEqual(call["function"]["arguments"], {"hour": [7, "x", 9]}) + + def test_already_decoded_array_is_not_cast_element_wise(self): + tools = _set_alarm_tools(groups={"type": "array", "items": {"type": "string"}}) + call = { + "type": "function", + "function": {"name": "set_alarm", "arguments": {"groups": ["[1,2]", "[]"]}}, + } + _parser_with_tools(tools)._coerce_tool_calls(call) + self.assertEqual(call["function"]["arguments"], {"groups": ["[1,2]", "[]"]}) + + def test_coerce_tool_calls_ignores_unusable_function_name(self): + # A transform can hand us a name parsed from model output, so a non-string name + # must be ignored rather than raising on the schema lookup. + parser = _parser_with_tools(_SET_ALARM_TOOLS) + call = {"type": "function", "function": {"name": ["set_alarm"], "arguments": {"hour": "7"}}} + self.assertEqual(parser._coerce_tool_calls(call), call) + self.assertEqual(call["function"]["arguments"], {"hour": "7"}) + + def test_union_with_container_still_keeps_scalar_text(self): + model_out = "\n\n\n1.50\n\n" + union = _set_alarm_tools(label={"anyOf": [{"type": "string"}, {"type": "object"}]}) + # `string` never casts, so the union keeps scalar-looking text as text ... + self.assertEqual( + _first_tool_args(parse_response(model_out, _XML_STRING_ARGS_TEMPLATE, prefix="", tools=union)), + {"label": "1.50"}, + ) + # ... while the `object` branch still decodes a body that really is an object. + object_body = '\n\n\n{"a": 1}\n\n' + self.assertEqual( + _first_tool_args(parse_response(object_body, _XML_STRING_ARGS_TEMPLATE, prefix="", tools=union)), + {"label": {"a": 1}}, + ) + + def test_tools_is_a_noop_for_typed_json_tool_calls(self): + # JSON tool-call bodies already carry their types, so passing tools= must not change them. + model_out = ( + "<|START_THINKING|>x<|END_THINKING|>" + '<|START_ACTION|>[{"tool_call_id": "0", "tool_name": "set_alarm", ' + '"parameters": {"hour": 7, "enabled": true}}]<|END_ACTION|><|END_OF_TURN_TOKEN|>' + ) + without = parse_response(model_out, cohere_template, prefix="") + with_tools = parse_response(model_out, cohere_template, prefix="", tools=_SET_ALARM_TOOLS) + self.assertEqual(without, with_tools) + self.assertEqual(_first_tool_args(with_tools), {"hour": 7, "enabled": True}) + + +if __name__ == "__main__": + unittest.main() +# fmt: on