diff --git a/tests/parser/test_chat_parsing.py b/tests/parser/test_chat_parsing.py index 26336a540a..679d97f15b 100644 --- a/tests/parser/test_chat_parsing.py +++ b/tests/parser/test_chat_parsing.py @@ -1160,6 +1160,111 @@ class ResponseEventStreamTest(unittest.TestCase): self.assertEqual(result, {"role": "assistant"}) self.assertEqual(final_events, []) + def test_region_events_expose_delimiter_offsets(self): + spec = { + "start_anchor": "", + "fields": { + "content": {"content_args": {"strip": False}}, + "tag": {"open_pattern": r"", "close": ""}, + }, + } + parser = ResponseParser(spec, prefix="") + events = parser.feed("plainbodytail") + events += parser.finalize()[1] + + boundaries = [ + (event["type"], event["field"], parser.input_text[event["start"] : event["end"]]) + for event in events + if event["type"] != "region_chunk" + ] + self.assertEqual( + boundaries, + [ + ("region_open", "content", ""), + ("region_close", "content", ""), + ("region_open", "tag", ""), + ("region_close", "tag", ""), + ("region_open", "content", ""), + ("region_close", "content", ""), + ], + ) + + def test_open_captures_and_prefix_boundary(self): + spec = { + "start_anchor": "[BEGIN]", + "fields": { + "tool_calls": { + "open_pattern": r"\w+)>", + "close": "", + "content": "json", + "transform": {"name": "{name}", "arguments": "{content}"}, + } + }, + } + parser = ResponseParser(spec, prefix="history[BEGIN]{"city":"Paris"}') + opening = events[0] + + self.assertEqual(prefix_end, len(""))) + self.assertEqual(opening["captures"], {"name": "get_weather"}) + + def test_held_prefix_bytes_are_not_rechunked(self): + spec = {"start_anchor": "[BEGIN]", "fields": {"content": {}, "tag": {"open": "", "close": ""}}} + parser = ResponseParser(spec, prefix="[BEGIN]Existing <") + chunks = [event["text"] for event in parser.feed("b> tail") if event["type"] == "region_chunk"] + message, _ = parser.finalize() + + self.assertEqual(chunks, ["b> tail"]) + self.assertEqual(message["content"], "Existing tail") + + def test_malformed_region_events_recover_and_continue(self): + spec = { + "start_anchor": "[BEGIN]", + "fields": { + "tool_calls": { + "open_pattern": r"\w+)>", + "close": "", + "content": "json", + "repeats": True, + "transform": { + "type": "function", + "function": {"name": "{name}", "arguments": "{content}"}, + }, + } + }, + } + parser = ResponseParser(spec, prefix="") + events = parser.feed('{"x":{"x":1}') + message, _ = parser.finalize() + opening, malformed = events[0], events[2] + + self.assertEqual(malformed["type"], "region_malformed") + self.assertIsInstance(malformed["error"], ValueError) + self.assertEqual(parser.input_text[opening["end"] : malformed["start"]], '{"x":') + self.assertEqual(parser.input_text[malformed["start"] : malformed["end"]], "") + self.assertEqual( + message["tool_calls"], + [{"type": "function", "function": {"name": "good", "arguments": {"x": 1}}}], + ) + + parser = ResponseParser(spec, prefix="") + parser.feed('{"x":') + _, events = parser.finalize() + self.assertEqual(events[-1]["type"], "region_malformed") + self.assertEqual(events[-1]["start"], len(parser.input_text)) + + def test_parse_response_raises_for_malformed_region(self): + spec = { + "start_anchor": "<|assistant|>", + "fields": {"x": {"open": "", "close": "", "content": "json"}}, + } + with self.assertRaisesRegex(ValueError, "could not parse region as JSON"): + parse_response('{"name":', spec, prefix="") + with self.assertRaisesRegex(ValueError, "could not parse region as JSON"): + parse_response("", spec, prefix='<|assistant|>{"name":') + class PrefixAndTruncationTest(unittest.TestCase): def test_prefix_lands_inside_explicit_region(self): diff --git a/vllm/parser/chat_parsing/response_parser.py b/vllm/parser/chat_parsing/response_parser.py index b739572759..556df803a7 100644 --- a/vllm/parser/chat_parsing/response_parser.py +++ b/vllm/parser/chat_parsing/response_parser.py @@ -85,8 +85,11 @@ def parse_response( """ response_template = load_response_template(response_template) stream = ResponseParser(response_template, prefix=prefix, tools=tools) - stream.feed(text) - message, _ = stream.finalize() + events = stream.feed(text) + message, final_events = stream.finalize() + for event in stream.initial_events + events + final_events: + if event["type"] == "region_malformed": + raise event["error"] return message @@ -110,7 +113,12 @@ class ResponseParser: 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". + Events can be "region_open", "region_chunk", "region_close", or "region_malformed". + Open, close and malformed events carry `start` and `end`, the span of `input_text` the + event consumed: its opening or closing delimiter, empty for implicit boundaries. + Explicit opens also carry the opener's named `captures`. A region whose value fails to + parse ends with "region_malformed" instead of "region_close", carrying the `error`, and + parsing continues after it. 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 @@ -155,10 +163,18 @@ class ResponseParser: self._body: str = "" self._opened: bool = False self._finalized: bool = False + self._prefix_len: int = 0 + self._malformed_fields: set[str] = set() self.initial_events: list[dict] = [] if prefix: self._consume_prefix(prefix) + @property + def input_text(self) -> str: + """Raw parser input after start-anchor truncation. Before the first `feed()`, this is + exactly the truncated prefix, so its length marks where generated text begins.""" + return self._buffer + 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. @@ -172,6 +188,7 @@ class ResponseParser: return self._buffer = truncated self._process(self.initial_events, eos=False) + self._prefix_len = len(truncated) def feed(self, text: str) -> list[dict]: """Feeds more text/tokens from the model output into the tokenizer, and returns any events that result @@ -197,7 +214,11 @@ class ResponseParser: 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] + missing = [ + n + for n, f in self._spec.fields.items() + if not f.optional and n not in self._output and n not in self._malformed_fields + ] if missing: raise ValueError(f"Required response_template fields missing from parsed output: {missing}") defaults = self._spec.defaults @@ -216,12 +237,12 @@ class ResponseParser: self._accumulate(events, self._buffer[self._pos : m.start()]) self._pos = m.end() if kind == "open": - self._close_current(events) + self._close_current(events, m.start(), m.start()) 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) + self._close_current(events, m.start(), m.end()) # 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(): @@ -233,7 +254,7 @@ class ResponseParser: if self._pos < len(self._buffer): self._accumulate(events, self._buffer[self._pos :]) self._pos = len(self._buffer) - self._close_current(events) + self._close_current(events, self._pos, self._pos) break # Stream everything up to the earliest still-pending delimiter. When # nothing is pending `hold_start == len(self._buffer)`, so this flushes @@ -341,20 +362,25 @@ class ResponseParser: return field = self._spec.fields[self._current] if not self._opened: - events.append({"type": "region_open", "field": self._current}) + events.append({"type": "region_open", "field": self._current, "start": self._pos, "end": self._pos}) 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}) + # Prefix bytes held back as a possible delimiter belong to the prompt, not the chunk. + text = text[max(0, self._prefix_len - self._pos) :] + if 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}) + events.append( + {"type": "region_open", "field": field.name, "start": m.start(), "end": m.end(), "captures": self._captures} + ) - def _close_current(self, events: list[dict]) -> None: + def _close_current(self, events: list[dict], close_start: int, close_end: int) -> 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.""" @@ -362,22 +388,30 @@ class ResponseParser: 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): + try: + value = process_field(self._body, field, self._captures) + if self._tool_params: + value = self._coerce_tool_calls(value) + if field.join is not None and not isinstance(value, str): raise ValueError( f"Field '{field.name}': 'join' requires each match to parse to a string, " f"got {type(value).__name__}." ) + except (KeyError, TypeError, ValueError) as error: + self._malformed_fields.add(self._current) + events.append( + {"type": "region_malformed", "field": self._current, "start": close_start, "end": close_end, "error": error} + ) + self._reset_to_implicit() + return + if field.join is not None: 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}) + events.append({"type": "region_close", "field": self._current, "start": close_start, "end": close_end, "value": value}) self._reset_to_implicit() def _coerce_tool_calls(self, value: Any) -> Any: