| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | # Copyright 2026 The HuggingFace Team. All rights reserved. |
| 5 | # ruff: noqa |
| 6 | # fmt: off |
| 7 | # |
| 8 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 9 | # you may not use this file except in compliance with the License. |
| 10 | # You may obtain a copy of the License at |
| 11 | # |
| 12 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 13 | # |
| 14 | # Unless required by applicable law or agreed to in writing, software |
| 15 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 16 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 17 | # See the License for the specific language governing permissions and |
| 18 | # limitations under the License. |
| 19 | |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import copy |
| 23 | import json |
| 24 | from typing import Any |
| 25 | |
| 26 | from .content_parsers import STREAMABLE_PARSERS, process_field |
| 27 | from .response_templates import ( |
| 28 | ResponseTemplate, |
| 29 | ResponseTemplateField, |
| 30 | load_response_template, |
| 31 | ) |
| 32 | |
| 33 | |
| 34 | def _schema_types(schema: Any) -> tuple[str, ...]: |
| 35 | if not isinstance(schema, dict): |
| 36 | return () |
| 37 | declared = schema.get("type") |
| 38 | types = [declared] if isinstance(declared, str) else [] |
| 39 | if isinstance(declared, list): |
| 40 | types.extend(t for t in declared if isinstance(t, str)) |
| 41 | for union_name in ("anyOf", "oneOf"): |
| 42 | for choice in schema.get(union_name) or []: |
| 43 | types.extend(_schema_types(choice)) |
| 44 | if schema.get("nullable") and "null" not in types: # `nullable` is how get_json_schema marks Optionals |
| 45 | types.append("null") |
| 46 | return tuple(types) |
| 47 | |
| 48 | |
| 49 | def _coerce(raw: str, types: tuple[str, ...]) -> Any: |
| 50 | for type_name in types: |
| 51 | try: |
| 52 | if type_name == "integer": |
| 53 | return int(raw) |
| 54 | if type_name == "number": |
| 55 | number = float(raw) |
| 56 | if number != number or number in (float("inf"), float("-inf")): |
| 57 | continue # NaN / inf are not valid JSON numbers |
| 58 | # Preserve ints when the source text had no fractional part. |
| 59 | return int(number) if number.is_integer() and "." not in raw else number |
| 60 | if type_name == "boolean" and raw.strip().lower() in ("true", "1", "false", "0"): |
| 61 | return raw.strip().lower() in ("true", "1") |
| 62 | if type_name == "null" and raw.strip() in ("null", "None"): |
| 63 | return None |
| 64 | if type_name in ("object", "array"): |
| 65 | decoded = json.loads(raw) |
| 66 | if isinstance(decoded, dict if type_name == "object" else list): |
| 67 | return decoded |
| 68 | except ValueError: |
| 69 | continue |
| 70 | return raw # `string` params, unknown types and failed casts all keep the original text |
| 71 | |
| 72 | |
| 73 | def parse_response( |
| 74 | text: str, |
| 75 | response_template: dict | ResponseTemplate, |
| 76 | *, |
| 77 | prefix: str | None = None, |
| 78 | tools: list[dict] | None = None, |
| 79 | ) -> dict: |
| 80 | """The main function for response parsing when you don't want streaming. Takes generated output |
| 81 | and the prompt prefix and parses them without streaming any events, then returns the parsed message. |
| 82 | |
| 83 | Pass OpenAI-style `tools` dictionaries to cast tool-call arguments |
| 84 | using the calling tool's JSON schema. |
| 85 | """ |
| 86 | response_template = load_response_template(response_template) |
| 87 | stream = ResponseParser(response_template, prefix=prefix, tools=tools) |
| 88 | events = stream.feed(text) |
| 89 | message, final_events = stream.finalize() |
| 90 | for event in stream.initial_events + events + final_events: |
| 91 | if event["type"] == "region_malformed": |
| 92 | raise event["error"] |
| 93 | return message |
| 94 | |
| 95 | |
| 96 | class ResponseParser: |
| 97 | """This class implements a streaming parser with a `response_template`. If you don't need streaming and |
| 98 | just want to parse a complete message, use the `parse_response` function above. Streaming parsing emits |
| 99 | events indicating when regions (message fields) are opened and closed, with the model writing to the region |
| 100 | that is currently open. |
| 101 | |
| 102 | Usage: |
| 103 | parser = ResponseParser(response_template, prefix=chat_prompt) |
| 104 | for event in parser.initial_events: |
| 105 | handle(event) |
| 106 | for chunk in model_text_stream: |
| 107 | for event in parser.feed(chunk): |
| 108 | handle(event) |
| 109 | message, final_events = parser.finalize() |
| 110 | for event in final_events: |
| 111 | handle(event) |
| 112 | |
| 113 | Pass OpenAI-style `tools=` dictionaries to cast tool-call arguments |
| 114 | using the calling tool's JSON schema as each region closes. |
| 115 | |
| 116 | Events can be "region_open", "region_chunk", "region_close", or "region_malformed". |
| 117 | Open, close and malformed events carry `start` and `end`, the span of `input_text` the |
| 118 | event consumed: its opening or closing delimiter, empty for implicit boundaries. |
| 119 | Explicit opens also carry the opener's named `captures`. A region whose value fails to |
| 120 | parse ends with "region_malformed" instead of "region_close", carrying the `error`, and |
| 121 | parsing continues after it. |
| 122 | |
| 123 | ResponseParser requires the chat `prefix` (i.e. the chat history, the prefill before the current generation). |
| 124 | This is because chat templates or assistant prefills can sometimes write part of the message, and if we |
| 125 | only see the model output, and not the template, then we can't reliably parse the message in those cases. |
| 126 | Any events produced while consuming the prefix are exposed as `initial_events`, so renderers can show |
| 127 | prefill regions before the model writes anything; closed prefill regions also land in the output dict. |
| 128 | """ |
| 129 | |
| 130 | def __init__( |
| 131 | self, |
| 132 | response_template: dict | ResponseTemplate, |
| 133 | prefix: str | None = None, |
| 134 | *, |
| 135 | tools: list[dict] | None = None, |
| 136 | ): |
| 137 | self._spec = load_response_template(response_template) |
| 138 | if prefix is None: |
| 139 | raise ValueError( |
| 140 | "`ResponseParser`/`parse_response` requires `prefix` (the chat prompt sent to the model before " |
| 141 | "generation), because chat templates often pre-write part of the assistant message (e.g. an " |
| 142 | "opening `<think>` tag) that the parser must see to parse the output correctly. If the generation " |
| 143 | 'already contains the complete message, pass `prefix=""` to opt out explicitly.' |
| 144 | ) |
| 145 | # Maps tool name -> schema `properties`, used to cast parsed tool-call arguments |
| 146 | self._tool_params: dict[str, dict] = {} |
| 147 | for tool in tools or []: |
| 148 | fn = tool.get("function", tool) if isinstance(tool, dict) else None |
| 149 | if isinstance(fn, dict) and isinstance(fn.get("name"), str): |
| 150 | parameters = fn.get("parameters") |
| 151 | properties = parameters.get("properties") if isinstance(parameters, dict) else None |
| 152 | self._tool_params[fn["name"]] = properties if isinstance(properties, dict) else {} |
| 153 | self._buffer: str = "" |
| 154 | self._pos: int = 0 |
| 155 | self._output: dict[str, Any] = copy.deepcopy(self._spec.defaults) |
| 156 | self._implicit_name: str | None = self._spec.implicit |
| 157 | # Unified current-region state: starts in the implicit region (or a |
| 158 | # null sink if none was declared), and returns there after every close. |
| 159 | # For explicit regions `_opened` flips to True eagerly on the open |
| 160 | # match; for the implicit region it flips lazily on the first byte. |
| 161 | self._current: str | None = self._implicit_name |
| 162 | self._captures: dict[str, str] = {} |
| 163 | self._body: str = "" |
| 164 | self._opened: bool = False |
| 165 | self._finalized: bool = False |
| 166 | self._prefix_len: int = 0 |
| 167 | self._malformed_fields: set[str] = set() |
| 168 | self.initial_events: list[dict] = [] |
| 169 | if prefix: |
| 170 | self._consume_prefix(prefix) |
| 171 | |
| 172 | @property |
| 173 | def input_text(self) -> str: |
| 174 | """Raw parser input after start-anchor truncation. Before the first `feed()`, this is |
| 175 | exactly the truncated prefix, so its length marks where generated text begins.""" |
| 176 | return self._buffer |
| 177 | |
| 178 | def _consume_prefix(self, prefix: str) -> None: |
| 179 | """Loads the prefix (the chat prefill sent to the model), right-truncates it to the start of the |
| 180 | assistant message (as determined by start_anchor) and then runs the remainder through the parser. |
| 181 | Events produced while processing the prefix are stashed on `initial_events` so callers can replay |
| 182 | them into a renderer before feeding model output. |
| 183 | |
| 184 | Think of this as the "get the parser up to speed on the story so far" method. |
| 185 | """ |
| 186 | truncated = self._spec.truncate_past_last_anchor(prefix) |
| 187 | if not truncated: |
| 188 | return |
| 189 | self._buffer = truncated |
| 190 | self._process(self.initial_events, eos=False) |
| 191 | self._prefix_len = len(truncated) |
| 192 | |
| 193 | def feed(self, text: str) -> list[dict]: |
| 194 | """Feeds more text/tokens from the model output into the tokenizer, and returns any events that result |
| 195 | (regions entered or left). This is the method you want to call after each generation step.""" |
| 196 | if self._finalized: |
| 197 | raise RuntimeError("ResponseParser already finalized") |
| 198 | if text: |
| 199 | self._buffer += text |
| 200 | events: list[dict] = [] |
| 201 | self._process(events, eos=False) |
| 202 | return events |
| 203 | |
| 204 | def finalize(self) -> tuple[dict, list[dict]]: |
| 205 | """Close the stream and return the final message dict together with |
| 206 | any finalization events. This is necessary because some regions may only |
| 207 | end at the end of the sequence, so you won't see the event telling you they're |
| 208 | ready until the sequence is finalized.""" |
| 209 | |
| 210 | def _is_empty(v: Any) -> bool: |
| 211 | return v is None or (isinstance(v, (list, dict, str)) and not v) |
| 212 | |
| 213 | if self._finalized: |
| 214 | raise RuntimeError("ResponseParser already finalized") |
| 215 | events: list[dict] = [] |
| 216 | self._process(events, eos=True) |
| 217 | missing = [ |
| 218 | n |
| 219 | for n, f in self._spec.fields.items() |
| 220 | if not f.optional and n not in self._output and n not in self._malformed_fields |
| 221 | ] |
| 222 | if missing: |
| 223 | raise ValueError(f"Required response_template fields missing from parsed output: {missing}") |
| 224 | defaults = self._spec.defaults |
| 225 | self._output = {k: v for k, v in self._output.items() if k in defaults or not _is_empty(v)} |
| 226 | self._finalized = True |
| 227 | return self._output, events |
| 228 | |
| 229 | def _process(self, events: list[dict], eos: bool) -> None: |
| 230 | while True: |
| 231 | watch = self._watchlist() |
| 232 | best, hold_start = self._scan(watch, eos) |
| 233 | |
| 234 | if best is not None: |
| 235 | kind, field, m = best |
| 236 | if m.start() > self._pos: |
| 237 | self._accumulate(events, self._buffer[self._pos : m.start()]) |
| 238 | self._pos = m.end() |
| 239 | if kind == "open": |
| 240 | self._close_current(events, m.start(), m.start()) |
| 241 | self._open_explicit(events, field, m) |
| 242 | else: # "close" (always the implicit region's close here, |
| 243 | # since explicit regions only expose their own close) |
| 244 | had_content = self._opened |
| 245 | self._close_current(events, m.start(), m.end()) |
| 246 | # Zero-width close on an already-empty region would just |
| 247 | # re-fire next iteration -- bail out to make progress. |
| 248 | if not had_content and m.start() == m.end(): |
| 249 | break |
| 250 | continue |
| 251 | |
| 252 | # No committable match in the current buffer. |
| 253 | if eos: |
| 254 | if self._pos < len(self._buffer): |
| 255 | self._accumulate(events, self._buffer[self._pos :]) |
| 256 | self._pos = len(self._buffer) |
| 257 | self._close_current(events, self._pos, self._pos) |
| 258 | break |
| 259 | # Stream everything up to the earliest still-pending delimiter. When |
| 260 | # nothing is pending `hold_start == len(self._buffer)`, so this flushes |
| 261 | # the whole buffer; otherwise we hold the (possibly partial) delimiter |
| 262 | # bytes back until more input resolves them. |
| 263 | if hold_start > self._pos: |
| 264 | self._accumulate(events, self._buffer[self._pos : hold_start]) |
| 265 | self._pos = hold_start |
| 266 | break |
| 267 | |
| 268 | def _watchlist(self) -> list[tuple[str, ResponseTemplateField]]: |
| 269 | """Patterns we care about right now: the close of the currently-open |
| 270 | explicit region, or -- if we're in the implicit/null region -- every |
| 271 | explicit open plus the implicit's own close (if any).""" |
| 272 | if self._current is not None and self._current != self._implicit_name: |
| 273 | field = self._spec.fields[self._current] |
| 274 | return [("close", field)] if field.close_re is not None else [] |
| 275 | watch: list[tuple[str, ResponseTemplateField]] = [] |
| 276 | for field in self._spec.fields.values(): |
| 277 | if field.open_re is not None: |
| 278 | watch.append(("open", field)) |
| 279 | if self._implicit_name is not None: |
| 280 | impl = self._spec.fields[self._implicit_name] |
| 281 | if impl.close_re is not None: |
| 282 | watch.append(("close", impl)) |
| 283 | return watch |
| 284 | |
| 285 | def _scan( |
| 286 | self, watch: list[tuple[str, ResponseTemplateField]], eos: bool |
| 287 | ) -> tuple[tuple[str, ResponseTemplateField, Any] | None, int]: |
| 288 | """Single pass over the watched delimiters, using the `regex` module's |
| 289 | partial matching to decide -- per delimiter -- whether it can be committed |
| 290 | now or must be held. Returns `(best, hold_start)`: |
| 291 | |
| 292 | * `best` is the earliest-starting delimiter we can safely commit *now* |
| 293 | (longest on ties, opens before closes), or `None`. |
| 294 | * `hold_start` is the leftmost buffer position occupied by a still-pending |
| 295 | match: a partial (incomplete) delimiter, or a complete one ending at the |
| 296 | buffer edge that more input could still grow. Bytes before it are safe to |
| 297 | emit; bytes from it onward must be held. It stays `len(self._buffer)` when |
| 298 | nothing is pending, letting the caller flush the whole buffer. |
| 299 | |
| 300 | A complete match is committable only if it starts strictly before |
| 301 | `hold_start` -- otherwise an earlier (or co-located) pending delimiter could |
| 302 | turn out to be the real one. At EOS nothing can grow, so partial matching is |
| 303 | skipped and every complete match is committable. |
| 304 | |
| 305 | (The `regex` module always reports the empty string as a live prefix, so a |
| 306 | partial search with no real match returns a zero-width match at the buffer |
| 307 | end; that lands in the pending branch with `start == len(self._buffer)`, a |
| 308 | no-op for `hold_start`.) |
| 309 | """ |
| 310 | best_key: tuple | None = None |
| 311 | best: tuple[str, ResponseTemplateField, Any] | None = None |
| 312 | hold_start = len(self._buffer) |
| 313 | for kind, field in watch: |
| 314 | # The watchlist only includes fields whose delimiter regex is set, so pattern is never None here. |
| 315 | pattern = field.open_re if kind == "open" else field.close_re |
| 316 | if eos: |
| 317 | m = pattern.search(self._buffer, self._pos) |
| 318 | else: |
| 319 | m = pattern.search(self._buffer, self._pos, partial=True) |
| 320 | if m is None: |
| 321 | continue |
| 322 | if not eos and (m.partial or self._can_grow(kind, field, m)): |
| 323 | # Pending: can't commit, and blocks emitting from its start onward. |
| 324 | hold_start = min(hold_start, m.start()) |
| 325 | continue |
| 326 | key = (m.start(), -(m.end() - m.start()), 0 if kind == "open" else 1, field.name) |
| 327 | if best_key is None or key < best_key: |
| 328 | best_key, best = key, (kind, field, m) |
| 329 | # A committable match co-located with or after a pending one must wait too: |
| 330 | # the pending delimiter starts no later and might be the one that fires. |
| 331 | if best is not None and best[2].start() >= hold_start: |
| 332 | best = None |
| 333 | return best, hold_start |
| 334 | |
| 335 | def _can_grow(self, kind: str, field: ResponseTemplateField, m: Any) -> bool: |
| 336 | r"""Whether a *complete* match ending at the current buffer edge could still |
| 337 | change as more input arrives -- in which case we defer rather than commit. A |
| 338 | match ending before the edge has already seen its terminating byte and is |
| 339 | final. At the edge: zero-width matches (`$` / `\Z`) are only real at true |
| 340 | EOS; a fully-present literal that no other literal in its set extends cannot |
| 341 | grow (the fast path that keeps literal delimiters zero-latency); anything |
| 342 | else (regex delimiters, prefix-overlapping literal lists) might.""" |
| 343 | if m.end() != len(self._buffer): |
| 344 | return False |
| 345 | if m.start() == m.end(): |
| 346 | return True |
| 347 | literals, can_extend = ( |
| 348 | (field.open_literals, field.open_literal_can_extend) |
| 349 | if kind == "open" |
| 350 | else (field.close_literals, field.close_literal_can_extend) |
| 351 | ) |
| 352 | return literals is None or can_extend |
| 353 | |
| 354 | def _accumulate(self, events: list[dict], text: str) -> None: |
| 355 | """Route `text` into the currently active region. When the current |
| 356 | region is the null sink (no implicit declared, no explicit open), we |
| 357 | silently discard. Every routed chunk emits a `region_chunk` event so |
| 358 | consumers can render live; `dirty=True` flags chunks from structured |
| 359 | parsers (json, xml-inline, kv-lines) whose raw bytes will only be |
| 360 | parsed into the final value on close.""" |
| 361 | if not text or self._current is None: |
| 362 | return |
| 363 | field = self._spec.fields[self._current] |
| 364 | if not self._opened: |
| 365 | events.append({"type": "region_open", "field": self._current, "start": self._pos, "end": self._pos}) |
| 366 | self._opened = True |
| 367 | self._body += text |
| 368 | # Prefix bytes held back as a possible delimiter belong to the prompt, not the chunk. |
| 369 | text = text[max(0, self._prefix_len - self._pos) :] |
| 370 | if text: |
| 371 | dirty = field.content not in STREAMABLE_PARSERS |
| 372 | events.append({"type": "region_chunk", "field": self._current, "text": text, "dirty": dirty}) |
| 373 | |
| 374 | def _open_explicit(self, events: list[dict], field: ResponseTemplateField, m: Any) -> None: |
| 375 | self._current = field.name |
| 376 | self._captures = {k: v for k, v in m.groupdict().items() if v is not None} |
| 377 | self._body = "" |
| 378 | self._opened = True |
| 379 | events.append( |
| 380 | {"type": "region_open", "field": field.name, "start": m.start(), "end": m.end(), "captures": self._captures} |
| 381 | ) |
| 382 | |
| 383 | def _close_current(self, events: list[dict], close_start: int, close_end: int) -> None: |
| 384 | """Close the current region and reset to the implicit/null region. |
| 385 | Skipped (aside from the reset) when the current region never opened -- |
| 386 | avoids vacuous open/close pairs at every explicit boundary.""" |
| 387 | if self._current is None or not self._opened: |
| 388 | self._reset_to_implicit() |
| 389 | return |
| 390 | field = self._spec.fields[self._current] |
| 391 | try: |
| 392 | value = process_field(self._body, field, self._captures) |
| 393 | if self._tool_params: |
| 394 | value = self._coerce_tool_calls(value) |
| 395 | if field.join is not None and not isinstance(value, str): |
| 396 | raise ValueError( |
| 397 | f"Field '{field.name}': 'join' requires each match to parse to a string, " |
| 398 | f"got {type(value).__name__}." |
| 399 | ) |
| 400 | except (KeyError, TypeError, ValueError) as error: |
| 401 | self._malformed_fields.add(self._current) |
| 402 | events.append( |
| 403 | {"type": "region_malformed", "field": self._current, "start": close_start, "end": close_end, "error": error} |
| 404 | ) |
| 405 | self._reset_to_implicit() |
| 406 | return |
| 407 | if field.join is not None: |
| 408 | previous = self._output.get(self._current) |
| 409 | self._output[self._current] = value if previous is None else previous + field.join + value |
| 410 | elif field.repeats: |
| 411 | self._output.setdefault(self._current, []).append(value) |
| 412 | else: |
| 413 | self._output[self._current] = value |
| 414 | events.append({"type": "region_close", "field": self._current, "start": close_start, "end": close_end, "value": value}) |
| 415 | self._reset_to_implicit() |
| 416 | |
| 417 | def _coerce_tool_calls(self, value: Any) -> Any: |
| 418 | if isinstance(value, list): |
| 419 | return [self._coerce_tool_calls(item) for item in value] |
| 420 | fn = value.get("function") if isinstance(value, dict) else None |
| 421 | if not isinstance(fn, dict): |
| 422 | return value |
| 423 | name, arguments = fn.get("name"), fn.get("arguments") |
| 424 | if not isinstance(name, str) or not isinstance(arguments, dict): |
| 425 | return value |
| 426 | if properties := self._tool_params.get(name): |
| 427 | for key, argument in arguments.items(): |
| 428 | if key not in properties or not (types := _schema_types(properties[key])): |
| 429 | continue |
| 430 | if isinstance(argument, str): |
| 431 | arguments[key] = _coerce(argument, types) |
| 432 | elif isinstance(argument, list) and "array" not in types: |
| 433 | # duplicate keys collected by `merge_duplicates` |
| 434 | arguments[key] = [_coerce(item, types) if isinstance(item, str) else item for item in argument] |
| 435 | return value |
| 436 | |
| 437 | def _reset_to_implicit(self) -> None: |
| 438 | self._current = self._implicit_name |
| 439 | self._captures = {} |
| 440 | self._body = "" |
| 441 | self._opened = False |
| 442 | # fmt: on |