diff --git a/tests/parser/test_chat_parsing.py b/tests/parser/test_chat_parsing.py
new file mode 100644
index 0000000000..26336a540a
--- /dev/null
+++ b/tests/parser/test_chat_parsing.py
@@ -0,0 +1,1662 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+# 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 six 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 typing import Any
+
+from vllm.parser.chat_parsing import ResponseParser, parse_response
+from vllm.parser.chat_parsing.response_parser import _coerce, _schema_types
+from vllm.parser.chat_parsing.response_templates import load_response_template
+
+
+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: dict[str, Any] = {
+ "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,
+# re-used 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[ev["field"]] += ev["text"]
+ elif t == "region_close":
+ self.assertEqual(open_field, ev["field"], f"close for non-open region: {ev}")
+ close_values[ev["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: dict[str, Any] = {
+ "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: dict[str, Any] = {
+ "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_get_json_schema_dialect(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: dict[str, Any] = 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: dict[str, Any] = {
+ "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: dict[str, Any] = {
+ "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: dict[str, Any] = {
+ "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
diff --git a/vllm/parser/chat_parsing/__init__.py b/vllm/parser/chat_parsing/__init__.py
new file mode 100644
index 0000000000..fad8a287e6
--- /dev/null
+++ b/vllm/parser/chat_parsing/__init__.py
@@ -0,0 +1,26 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+# 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/vllm/parser/chat_parsing/content_parsers.py b/vllm/parser/chat_parsing/content_parsers.py
new file mode 100644
index 0000000000..21bd8447a9
--- /dev/null
+++ b/vllm/parser/chat_parsing/content_parsers.py
@@ -0,0 +1,255 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+# 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/vllm/parser/chat_parsing/response_parser.py b/vllm/parser/chat_parsing/response_parser.py
new file mode 100644
index 0000000000..b739572759
--- /dev/null
+++ b/vllm/parser/chat_parsing/response_parser.py
@@ -0,0 +1,408 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+# 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/vllm/parser/chat_parsing/response_templates.py b/vllm/parser/chat_parsing/response_templates.py
new file mode 100644
index 0000000000..2636e32f71
--- /dev/null
+++ b/vllm/parser/chat_parsing/response_templates.py
@@ -0,0 +1,249 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+
+# 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(
+ "response_template defines %s but the anchor was not found "
+ "in the prefix; the parser will process the entire prefix instead.",
+ kind,
+ )
+ 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