class MistralParser(ParserEngine):
"""Mistral parser: engine-based reasoning + ``[TOOL_CALLS]`` tool calls.
Reasoning encoding is auto-detected from the tokenizer:
- ``"special_token"`` – ``[THINK]`` present in vocab (v13+).
- ``"text"`` – tokenizer supports grammar but has no ``[THINK]`` (v11).
- ``"none"`` – no grammar support; reasoning disabled.
Tool calls use the ``[TOOL_CALLS]func_name{...}`` format. The opening
``{`` doubles as the NAME→ARGS separator and is included in the argument
buffer via an ``ARG_VALUE_CHUNK`` event on the transition (mirrors
kimi_k2, JSON args, `tool_args_json=True`).
When the tokenizer does not support grammar (``_reasoning_encoding ==
"none"``), the legacy ``[TOOL_CALLS]``-based extraction path is used
instead of the declarative engine, handling both pre-v11 JSON-array and
v11+ ``funcname{args}`` formats.
"""
def __init__(
self,
tokenizer: TokenizerLike,
tools: list[Tool] | None = None,
**kwargs,
) -> None:
vocab = tokenizer.get_vocab()
self._reasoning_encoding: Literal["special_token", "text", "none"]
if _THINK_START_SPECIAL in vocab:
self._reasoning_encoding = "special_token"
elif getattr(tokenizer, "supports_grammar", False):
self._reasoning_encoding = "text"
else:
self._reasoning_encoding = "none"
kwargs.setdefault(
"parser_engine_config",
mistral_config(reasoning_encoding=self._reasoning_encoding),
)
super().__init__(tokenizer, tools, **kwargs)
self._tool_calls_token_id: int | None = self.vocab.get(_TOOL_CALLS)
# Tool calls use the legacy parser for all tokenizer versions;
# reasoning is handled by the engine.
self.bot_token: str = _TOOL_CALLS
self.bot_token_id: int | None = self._tool_calls_token_id
if self.bot_token_id is None:
raise RuntimeError(
"Mistral parser could not locate the tool call token in the tokenizer!"
)
# Legacy tool-call streaming state.
self.prev_tool_call_arr: list[dict[str, Any]] = []
self.current_tool_id: int = -1
self.streaming_state: StreamingState = StreamingState.WAITING_FOR_TOOL_START
self.tool_call_started: bool = False
self.current_tool_name: str | None = None
self.current_tool_mistral_id: str | None = None
self.starting_new_tool: bool = False
self.streamed_args_for_tool: list[str] = []
self._is_pre_v11: bool = _is_pre_v11_tokeniser(tokenizer)
self.current_tool_string_arguments: str | None = None
self.parse_coro = None
if self._is_pre_v11:
self.parse_coro = ijson.parse_coro(
self.update_stream_state_pre_v11_tokenizer()
)
self.tool_call_regex = re.compile(r"\[{.*}\]", re.DOTALL)
def adjust_request(
self, request: ChatCompletionRequest | ResponsesRequest
) -> ChatCompletionRequest | ResponsesRequest:
if not isinstance(request, ResponsesRequest) and request._grammar_from_parser:
return request
so_non_supported_attributes = [
"regex",
"choice",
"grammar",
# whitespace_pattern is not a constraint type but an option;
# Mistral grammar factory does not support it.
"whitespace_pattern",
"structural_tag",
]
any_so_non_supported_active = request.structured_outputs is not None and any(
getattr(request.structured_outputs, attribute) is not None
for attribute in so_non_supported_attributes
)
response_format_non_supported_active = (
isinstance(request, ResponsesRequest)
or request.response_format is not None
and request.response_format.type == "structural_tag"
)
if (
not is_mistral_tokenizer(self.model_tokenizer)
or isinstance(request, ResponsesRequest)
or not self.model_tokenizer.supports_grammar
or any_so_non_supported_active
or response_format_non_supported_active
):
request = super().adjust_request(request)
if request.tools and request.tool_choice != "none":
# Keep special tokens so the [TOOL_CALLS] marker
# survives for tool detection.
request.skip_special_tokens = False
# Inject a guided JSON schema for pre-v11 required/named tool choice
# so the model emits a well-formed bare JSON array instead of
# rambling. tool_choice forces a tool call, so a json_object /
# json_schema response_format is cleared here (the tool schema is
# the sole structured-output constraint), mirroring the base
# ToolParser; a structural_tag response_format is left untouched.
req_tool_choice = request.tool_choice
is_required_or_named = req_tool_choice == "required" or isinstance(
req_tool_choice, ChatCompletionNamedToolChoiceParam
)
response_format = getattr(request, "response_format", None)
response_format_overridable = response_format is None or (
response_format.type in ("text", "json_object", "json_schema")
)
# This runs only in the non-grammar (legacy) branch, so it covers
# both pre-v11 Mistral tokenizers and non-Mistral (e.g. HF-mode)
# tokenizers driving the Mistral tool parser. The remaining guards
# keep it off when the user supplied their own structured output.
if (
not isinstance(request, ResponsesRequest)
and request.tools
and request.structured_outputs is None
and is_required_or_named
and response_format_overridable
):
schema = self._build_guided_schema_pre_v11(request)
if schema is not None:
request.structured_outputs = StructuredOutputsParams(json=schema)
request.response_format = None
return request
json_schema: dict[str, Any] | None = None
if request.structured_outputs is not None:
if request.structured_outputs.json_object is not None:
json_schema = _DEFAULT_JSON_SCHEMA
elif request.structured_outputs.json is not None:
if isinstance(request.structured_outputs.json, str):
check_json_nesting(request.structured_outputs.json)
json_schema = json.loads(request.structured_outputs.json)
else:
json_schema = request.structured_outputs.json
else:
raise ValueError(
"Unsupported request.structured_outputs for MistralParser. "
"Only `json` and `json_object` are supported."
)
elif (
request.response_format is not None
and request.response_format.type != "text"
):
if request.response_format.type == "json_object":
json_schema = _DEFAULT_JSON_SCHEMA
elif request.response_format.type == "json_schema":
if request.response_format.json_schema is not None:
json_schema = request.response_format.json_schema.json_schema
else:
json_schema = _DEFAULT_JSON_SCHEMA
else:
raise ValueError(
"MistralParser only accepts `text`, `json_object` or "
f"`json_schema`, got {request.response_format=}"
)
request.response_format = None
grammar_factory = self.model_tokenizer.grammar_factory
# Rendering grammar is cached in mistral-common given tools, template and mode.
template = grammar_factory.select_jinja_template()
mistral_tools = (
[MistralTool.from_openai(tool.model_dump()) for tool in request.tools]
if request.tools is not None
else None
)
tool_choice: MistralToolChoice
match request.tool_choice:
case "none" | "auto" | "required":
tool_choice = MistralToolChoiceEnum(request.tool_choice)
case None:
tool_choice = MistralToolChoiceEnum.auto
# _ == Named tool choice
case _:
tool_choice = MistralNamedToolChoice.model_validate(
{
"type": "function",
"function": {"name": request.tool_choice.function.name},
}
)
match tool_choice, json_schema is not None:
case MistralToolChoiceEnum.none, True:
lark_grammar = grammar_factory.get_lark_for_json_schema(
template=template, json_schema=json_schema
)
case _, _:
lark_grammar = grammar_factory.get_lark_from_jinja(
template=template,
mode=tool_choice,
tools=mistral_tools,
json_schema=json_schema,
parallel_tool_calls=request.parallel_tool_calls,
json_only=False,
)
request.structured_outputs = StructuredOutputsParams(grammar=lark_grammar)
request._grammar_from_parser = True
return request
def _build_guided_schema_pre_v11(
self,
request: ChatCompletionRequest,
) -> dict[str, Any] | None:
"""Build a guided JSON schema for pre-v11 required/named tool choice.
The schema enforces the Mistral-native array format
``[{"name": ..., "arguments": {...}}]`` so the model emits a parseable
bare JSON array instead of free-form text.
Args:
request: The chat completion request carrying `tools` and
`tool_choice`.
Returns:
A JSON Schema dict, or ``None`` if the named tool is not found.
"""
tool_choice = request.tool_choice
tools = request.tools or []
extra: dict[str, Any] = {}
if tool_choice == "required":
applicable_tools = tools
else:
# Named tool choice — restrict to the single requested tool.
assert isinstance(tool_choice, ChatCompletionNamedToolChoiceParam)
chosen_name = tool_choice.function.name
applicable_tools = [t for t in tools if t.function.name == chosen_name]
if not applicable_tools:
logger.warning(
"Named tool %r not found in tools list; "
"skipping guided schema injection.",
chosen_name,
)
return None
extra["maxItems"] = 1
any_of = [
{
"type": "object",
"properties": {
"name": {"type": "string", "enum": [tool.function.name]},
"arguments": tool.function.parameters or {"type": "object"},
},
"required": ["name", "arguments"],
}
for tool in applicable_tools
]
return {
"type": "array",
"minItems": 1,
**extra,
"items": {
"type": "object",
"anyOf": any_of,
},
}
def _ensure_tool_id(self, slot: ToolCallSlot, name: str) -> None:
"""Assign a Mistral-compatible 9-char alphanumeric id to `slot`."""
if not slot.id:
slot.id = MistralToolCall.generate_random_id()
def extract_tool_calls_from_content(
self,
content: str,
request: ChatCompletionRequest,
) -> ExtractedToolCallInformation:
if self._is_pre_v11:
return self._legacy_extract_tool_calls(content, request)
return super().extract_tool_calls_from_content(content, request)
def _legacy_extract_tool_calls(
self,
model_output: str,
request: ChatCompletionRequest | None,
) -> ExtractedToolCallInformation:
"""Pre-v11 non-streaming extraction.
Handles ``[TOOL_CALLS][{...}]`` and guided bare-array formats.
"""
if request is None:
tool_choice = None
tools = None
else:
tool_choice = request.tool_choice
tools = request.tools
# tool_choice="none" with tools: never produce tool calls.
if tool_choice == "none" and tools:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
content: str | None = None
if self.bot_token in model_output:
# pre-v11: content[BOT] [{tool_call1},{tool_call2}]
# Repeated BOT tokens with nothing between them count as one, and
# output after a later BOT token is trailing text, as in streaming.
content, *segments = model_output.split(self.bot_token)
stringified_tool_calls = next(
(segment.strip() for segment in segments if segment.strip()), ""
)
elif tool_choice == "required" or isinstance(
tool_choice, ChatCompletionNamedToolChoiceParam
):
# Guided bare-array output (no [TOOL_CALLS] marker).
stringified_tool_calls = model_output.strip()
else:
return ExtractedToolCallInformation(
tools_called=False, tool_calls=[], content=model_output
)
try:
# Use raw_decode to parse the first valid JSON value,
# ignoring trailing tokens the model may emit after
# the tool call array.
parsed, _ = json.JSONDecoder().raw_decode(stringified_tool_calls)
except json.JSONDecodeError:
try:
raw_tool_call = self.tool_call_regex.findall(stringified_tool_calls)[0]
parsed = json.loads(raw_tool_call)
except (IndexError, json.JSONDecodeError):
logger.exception("Error in extracting tool call from response.")
parsed = None
function_calls = _pre_v11_function_calls(parsed)
if not function_calls:
if parsed is not None:
logger.warning(
"No tool call found in the JSON after %s; returning it as content.",
self.bot_token,
)
return ExtractedToolCallInformation(
tools_called=False,
tool_calls=[],
content=stringified_tool_calls,
)
mistral_tool_calls: list[MistralToolCall] = [
MistralToolCall(type="function", function=function_call)
for function_call in function_calls
]
return ExtractedToolCallInformation(
tools_called=True,
tool_calls=mistral_tool_calls,
content=content if content and content.strip() else None,
)
def extract_tool_calls_streaming(
self,
previous_text: str,
current_text: str,
delta_text: str,
previous_token_ids: Sequence[int],
current_token_ids: Sequence[int],
delta_token_ids: Sequence[int],
request: ChatCompletionRequest | ResponsesRequest,
) -> DeltaMessage | None:
if not self._is_pre_v11:
return super().extract_tool_calls_streaming(
previous_text,
current_text,
delta_text,
previous_token_ids,
current_token_ids,
delta_token_ids,
request,
)
# Pre-v11: latch on [TOOL_CALLS] or on the first content of a guided
# required/named request (bare JSON array, no special token).
if self.bot_token_id in delta_token_ids or self.bot_token in delta_text:
self.tool_call_started = True
elif not self.tool_call_started and delta_text:
is_guided = request.tool_choice == "required" or isinstance(
request.tool_choice, ChatCompletionNamedToolChoiceParam
)
if is_guided:
self.tool_call_started = True
if not self.tool_call_started:
return DeltaMessage(content=delta_text)
try:
return self._extract_tool_calls_streaming_pre_v11_tokenizer(
delta_text=delta_text,
delta_token_ids=delta_token_ids,
)
except Exception:
logger.exception("Error trying to handle streaming tool call.")
return None
@ijson.coroutine
def update_stream_state_pre_v11_tokenizer(self):
bare_object = False
while True:
(prefix, event, value) = yield
# A bare top-level object is a single call: parse it as an item.
if prefix == "" and event == "start_map":
bare_object = True
if bare_object:
prefix = f"item.{prefix}" if prefix else "item"
if prefix == "item" and event == "start_map":
self.streaming_state = StreamingState.WAITING_FOR_TOOL_KEY
self.starting_new_tool = True
if prefix == "item" and event == "map_key" and value == "name":
self.streaming_state = StreamingState.PARSING_NAME
if prefix == "item.name" and event == "string":
self.current_tool_name = value
self.streaming_state = StreamingState.PARSING_NAME_COMPLETED
if prefix == "item" and event == "map_key" and value == "arguments":
self.streaming_state = StreamingState.WAITING_FOR_ARGUMENTS_START
if prefix == "item.arguments" and event == "start_map":
self.streaming_state = StreamingState.PARSING_ARGUMENTS
if prefix == "item.arguments" and event == "end_map":
self.streaming_state = StreamingState.PARSING_ARGUMENTS_COMPLETED
if prefix == "item.arguments" and event == "string":
self.current_tool_string_arguments = value
if prefix == "item" and event == "end_map":
self.streaming_state = (
StreamingState.ALL_TOOLS_COMPLETE
if bare_object
else StreamingState.TOOL_COMPLETE
)
if prefix == "" and event == "end_array":
self.streaming_state = StreamingState.ALL_TOOLS_COMPLETE
def _extract_tool_calls_streaming_pre_v11_tokenizer(
self,
delta_text: str,
delta_token_ids: Sequence[int],
) -> DeltaMessage | None:
"""Extract tool calls for pre-v11 Mistral models.
Handles ``[TOOL_CALLS][{"name": "add", "arguments":{"a": 3.5}}]``.
"""
assert self.parse_coro is not None
content = None
delta_tool_calls: list[DeltaToolCall] = []
current_tool_call: DeltaToolCall = DeltaToolCall(
index=self.current_tool_id, type="function"
)
current_tool_call_modified = False
if self.bot_token_id in delta_token_ids or self.bot_token in delta_text:
# this is the first tool call
if not delta_text.startswith(self.bot_token):
content = delta_text.split(self.bot_token)[0]
delta_text = "".join(delta_text.split(self.bot_token)[1:])
# ijson gives no text index per event, so split the delta manually
# to know where each event is emitted from.
while len(delta_text) > 0:
streaming_state_before_parse = self.streaming_state
if self.streaming_state == StreamingState.WAITING_FOR_TOOL_START:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_opening_curly_braces=1,
)
# Every state in which the call object can close stops after "}",
# so text following a bare object is never sent to ijson.
elif self.streaming_state == StreamingState.WAITING_FOR_TOOL_KEY:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_colon=1,
stop_after_opening_curly_braces=1,
stop_after_closing_curly_braces=1,
)
elif self.streaming_state == StreamingState.PARSING_NAME:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_comma=1,
stop_after_closing_brackets=1,
stop_after_closing_curly_braces=1,
)
elif self.streaming_state == StreamingState.WAITING_FOR_ARGUMENTS_START:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_opening_curly_braces=1,
stop_after_closing_curly_braces=1,
)
elif self.streaming_state == StreamingState.PARSING_ARGUMENTS:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_closing_curly_braces=1,
)
elif self.streaming_state in [
StreamingState.PARSING_ARGUMENTS_COMPLETED,
StreamingState.PARSING_NAME_COMPLETED,
]:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_closing_curly_braces=1,
stop_after_closing_brackets=1,
)
elif self.streaming_state == StreamingState.TOOL_COMPLETE:
delta_to_be_parsed, delta_text = self._split_delta(
delta_text=delta_text,
stop_after_opening_curly_braces=1,
stop_after_closing_brackets=1,
)
elif self.streaming_state == StreamingState.ALL_TOOLS_COMPLETE:
content = delta_text
delta_text = ""
else:
delta_to_be_parsed = delta_text
delta_text = ""
if self.streaming_state != StreamingState.ALL_TOOLS_COMPLETE:
self.parse_coro.send(delta_to_be_parsed.encode("utf-8"))
# start_map is the authoritative new-tool signal and survives
# batched deltas, unlike comparing pre/post streaming states.
if self.starting_new_tool:
self.starting_new_tool = False
if current_tool_call_modified:
if self.current_tool_mistral_id is not None:
current_tool_call.id = self.current_tool_mistral_id
self.current_tool_mistral_id = None
self._track_streamed_args_pre_v11(current_tool_call)
delta_tool_calls.append(current_tool_call)
current_tool_call_modified = False
self.current_tool_id += 1
self.streamed_args_for_tool.append("")
self.prev_tool_call_arr.append({})
self.current_tool_mistral_id = MistralToolCall.generate_random_id()
current_tool_call = DeltaToolCall(
index=self.current_tool_id,
type="function",
)
if current_tool_call.function is None:
current_tool_call.function = DeltaFunctionCall()
if self.current_tool_name is not None:
current_tool_call_modified = True
current_tool_call.function.name = self.current_tool_name
self.prev_tool_call_arr[self.current_tool_id]["name"] = (
self.current_tool_name
)
self.current_tool_name = None
if self.current_tool_string_arguments is not None:
current_tool_call_modified = True
current_tool_call.function.arguments = (
self.current_tool_string_arguments
)
self.current_tool_string_arguments = None
if self.streaming_state == StreamingState.PARSING_NAME_COMPLETED:
self.streaming_state = StreamingState.WAITING_FOR_TOOL_KEY
if self.streaming_state in [
StreamingState.PARSING_ARGUMENTS,
StreamingState.PARSING_ARGUMENTS_COMPLETED,
]:
if self.streaming_state == StreamingState.PARSING_ARGUMENTS_COMPLETED:
self.streaming_state = StreamingState.WAITING_FOR_TOOL_KEY
current_tool_call_modified = True
if current_tool_call.function.arguments is None:
current_tool_call.function.arguments = delta_to_be_parsed
else:
current_tool_call.function.arguments += delta_to_be_parsed
if streaming_state_before_parse != StreamingState.PARSING_ARGUMENTS:
# It's the first chunk of arg. let's lstrip it
current_tool_call.function.arguments = (
current_tool_call.function.arguments.lstrip()
)
if current_tool_call_modified:
if self.current_tool_mistral_id is not None:
current_tool_call.id = self.current_tool_mistral_id
self.current_tool_mistral_id = None
self._track_streamed_args_pre_v11(current_tool_call)
delta_tool_calls.append(current_tool_call)
if content or len(delta_tool_calls) > 0:
delta_message = DeltaMessage()
if content:
delta_message.content = content
if len(delta_tool_calls) > 0:
delta_message.tool_calls = delta_tool_calls
return delta_message
else:
if self.streaming_state == StreamingState.ALL_TOOLS_COMPLETE:
return DeltaMessage()
else:
return None
def _track_streamed_args_pre_v11(self, tool_call: DeltaToolCall) -> None:
r"""Accumulate `tool_call` arguments into the streaming state."""
if tool_call.function is not None and tool_call.function.arguments is not None:
self.streamed_args_for_tool[self.current_tool_id] += (
tool_call.function.arguments
)
self.prev_tool_call_arr[self.current_tool_id]["arguments"] = (
self.streamed_args_for_tool[self.current_tool_id]
)
def _split_delta(
self,
delta_text: str,
stop_after_quotes: int = -1,
stop_after_opening_curly_braces: int = -1,
stop_after_closing_curly_braces: int = -1,
stop_after_closing_brackets: int = -1,
stop_after_colon: int = -1,
stop_after_comma: int = -1,
) -> tuple[str, str]:
delta_to_be_parsed = ""
for i, c in enumerate(delta_text):
if c in ['"', "'"]:
delta_to_be_parsed += c
stop_after_quotes -= 1
if stop_after_quotes == 0:
return (delta_to_be_parsed, delta_text[i + 1 :])
elif c == "{":
delta_to_be_parsed += c
stop_after_opening_curly_braces -= 1
if stop_after_opening_curly_braces == 0:
return (delta_to_be_parsed, delta_text[i + 1 :])
elif c == "}":
delta_to_be_parsed += c
stop_after_closing_curly_braces -= 1
if stop_after_closing_curly_braces == 0:
return (delta_to_be_parsed, delta_text[i + 1 :])
elif c == "]":
delta_to_be_parsed += c
stop_after_closing_brackets -= 1
if stop_after_closing_brackets == 0:
return (delta_to_be_parsed, delta_text[i + 1 :])
elif c == ":":
delta_to_be_parsed += c
stop_after_colon -= 1
if stop_after_colon == 0:
return (delta_to_be_parsed, delta_text[i + 1 :])
elif c == ",":
delta_to_be_parsed += c
stop_after_comma -= 1
if stop_after_comma == 0:
return (delta_to_be_parsed, delta_text[i + 1 :])
else:
delta_to_be_parsed += c
return (delta_to_be_parsed, "")
def extract_reasoning(
self,
model_output: str,
request: ChatCompletionRequest | ResponsesRequest,
) -> tuple[str | None, str | None]:
if self._reasoning_encoding == "none":
return None, model_output
return super().extract_reasoning(model_output, request)
def _accept_tool_name(self, name: str) -> bool:
# Once `[ARGS]` or the opening `{` has moved the slot past its name,
# that name is final, empty included. Emitting the call with ""
# surfaces the malformed generation instead of dropping it.
return self._is_valid_tool_name(name)
def _try_extract_name(self, idx: int) -> str | None:
# No JSON-embedded "name" key in this format; the slot name is final.
return self._tool_slots[idx].name
def _extract_name_and_args(self, raw_body: str) -> tuple[str, str]:
# Never a {"name": ..., "arguments": ...} envelope -- a literal "name"
# argument key must not be mistaken for the tool name.
return "", self._extract_args_json(raw_body, "")
def _handle_arg_chunk(
self,
event: SemanticEvent,
deltas: list[DeltaToolCall],
) -> None:
"""Emit the opening ``{`` as an arg delta when the name is first sent.
When the TOOL_NAME→TOOL_ARGS transition fires, ``{`` arrives as an
ARG_VALUE_CHUNK before ``name_sent`` is True. The parent emits the
name delta but not the ``{`` arg delta. This override re-emits the
current chunk so streaming clients receive a valid JSON prefix.
"""
idx = event.tool_index
name_sent_before = (
0 <= idx < len(self._tool_slots) and self._tool_slots[idx].name_sent
)
super()._handle_arg_chunk(event, deltas)
if (
event.value
and not name_sent_before
and 0 <= idx < len(self._tool_slots)
and self._tool_slots[idx].name_sent
):
deltas.append(
DeltaToolCall(
index=idx,
function=DeltaFunctionCall(arguments=event.value),
)
)
def _extract_args_json(self, raw_args: str, func_name: str) -> str:
"""Return the first complete JSON value in ``raw_args``.
v11+ tool calls are emitted as ``name{args}`` with no terminator, so a
model may append ordinary text after the closing brace. Parse the first
JSON value with ``raw_decode`` and drop any trailing output, mirroring
the pre-v11 path (fixes gh#48975).
"""
stripped = raw_args.strip()
if not stripped:
return "{}"
try:
_, end = json.JSONDecoder().raw_decode(stripped)
except json.JSONDecodeError:
return stripped
return stripped[:end]