solar_open_tool_parser.py

# coding=utf-8
# Copyright 2025 Upstage AI.
#
# 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.

import random
import re
import string
import ast
import json
from collections.abc import Sequence
from typing import Union, Tuple, List, Optional

from vllm.entrypoints.openai.protocol import (
    ChatCompletionRequest,
    DeltaMessage,
    DeltaFunctionCall,
    DeltaToolCall,
    ExtractedToolCallInformation,
    ToolCall,
    FunctionCall,
)
from vllm.entrypoints.openai.tool_parsers.abstract_tool_parser import (
    ToolParser
)
from vllm.logger import init_logger

import pyjson5

class ToolCallID:
    _LENGTH = 10

    def __init__(self, id_val: str, validation: bool = False):
        self._id = id_val
        if validation:
            self._validate()

    @classmethod
    def random(cls, validation=False) -> 'ToolCallID':
        chars = string.ascii_lowercase + string.digits
        return cls(''.join(random.choice(chars) for _ in range(ToolCallID._LENGTH)), validation=validation)

    def _validate(self):
        assert len(self._id) == ToolCallID._LENGTH
        pattern = r'^[a-z0-9]{10}$'
        assert re.match(pattern, self._id) is not None

    def to_string(self) -> str:
        return self._id

    def __str__(self) -> str:
        return self.to_string()


logger = init_logger(__name__)


class SolarOpenToolParser(ToolParser):

    def extract_tool_calls(
            self,
            model_output: str,
            request: ChatCompletionRequest,
    ) -> ExtractedToolCallInformation:
        content, tool_calls = self._parse_text(model_output)
        return ExtractedToolCallInformation(
            tools_called=len(tool_calls) > 0,
            tool_calls=tool_calls,
            content=content if content 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,
    ) -> Union[DeltaMessage, None]:
        # 1) Emit plain content tokens immediately until content terminator
        # tags or tool_calls section begins. Be careful when tokenizer groups
        # multiple special tags into a single delta (e.g., "<|tool_calls|><|tool_call:begin|>").
        # Only emit as content if BOTH:
        #  - previous_text has not seen any special markers, and
        #  - delta_text does NOT contain any of those markers as a substring.
        if delta_text:
            # Do NOT emit content if we have already started any special section
            # including tool call tags. Content should only be emitted at the
            # very beginning before any markers show up.
            special_markers = (
                "<|flush|>",
                "<|end|>",
                "<|begin|>",
                "<|tool_calls|>",
                "<|tool_call:begin|>",
                "<|tool_call:name|>",
                "<|tool_call:args|>",
                "<|tool_call:end|>",
                "<|calls|>",
            )
            if not any(tag in previous_text for tag in special_markers):
                if not any(tag in delta_text for tag in special_markers):
                    return DeltaMessage(content=delta_text, tool_calls=[])

        tool_call_deltas: list[DeltaToolCall] = []

        # Helper lambdas to analyze current_text state
        def _completed_calls_count(txt: str) -> int:
            return len(self._parse_tool_calls(txt))

        # Detect if a new tool_call started streaming its args just now.
        if delta_text and "<|tool_call:args|>" in delta_text:
            # Extract id and name for the latest tool call block present so far.
            begin_tag = "<|tool_call:begin|>"
            name_tag = "<|tool_call:name|>"
            args_tag = "<|tool_call:args|>"

            latest_args = current_text.rfind(args_tag)
            latest_name = current_text.rfind(name_tag, 0, latest_args if latest_args != -1 else None)
            latest_begin = current_text.rfind(begin_tag, 0, latest_name if latest_name != -1 else None)
            if latest_begin != -1 and latest_name != -1 and latest_args != -1 and latest_begin < latest_name < latest_args:
                tool_id = current_text[latest_begin + len(begin_tag):latest_name]
                func_name = current_text[latest_name + len(name_tag):latest_args]
                # Index equals number of args tags seen before this delta
                index = previous_text.count(args_tag)
                tool_call_deltas.append(
                    DeltaToolCall(
                        id=tool_id,
                        type="function",
                        index=index,
                        function=DeltaFunctionCall(name=func_name, arguments=""),
                    )
                )

        # If we are inside args (after last args tag without end), stream arg chunk
        begin_tag = "<|tool_call:begin|>"
        args_tag = "<|tool_call:args|>"
        end_tag = "<|tool_call:end|>"
        last_args_pos = current_text.rfind(args_tag)
        last_end_pos = current_text.rfind(end_tag)
        if last_args_pos != -1 and (last_end_pos == -1 or last_args_pos > last_end_pos):
            # Currently within args for the latest tool call
            # Determine previous args text and current args text to compute delta
            prev_last_args = previous_text.rfind(args_tag)
            prev_last_end = previous_text.rfind(end_tag)
            if prev_last_args != -1 and (prev_last_end == -1 or prev_last_args > prev_last_end):
                # Already inside args previously: emit only the delta_text
                if delta_text and delta_text not in (begin_tag, args_tag, end_tag):
                    # Stream into the most recently started (but not yet ended) call
                    index = max(previous_text.count(args_tag) - 1, 0)
                    tool_call_deltas.append(
                        DeltaToolCall(
                            id=None,
                            type=None,
                            index=index,
                            function=DeltaFunctionCall(name=None, arguments=delta_text),
                        )
                    )

        if not tool_call_deltas:
            return None

        return DeltaMessage(content=None, tool_calls=tool_call_deltas)

    # --------------------
    # Internal helpers
    # --------------------
    def _parse_text(self, text: str) -> Tuple[Optional[str], List[ToolCall]]:
        """Parse the completed segments from the given text.

        Returns (content, tool_calls) where content is extracted as the leading
        text up to the first '<|flush|>' or '<|end|>' marker, and tool_calls is
        a list of fully parsed tool calls inside '<|tool_calls|> ... <|calls|>'.
        """
        content = self._parse_content(text)
        tool_calls = self._parse_tool_calls(text)
        return content, tool_calls

    def _parse_content(self, text: str) -> Optional[str]:
        """Extract assistant content from the text.

        Rule: take the leading content before the first '<|flush|>' or
        '<|end|>' marker. If neither marker exists, return None.
        """
        end_tags = ["<|flush|>", "<|end|>"]

        # Take leading content before the first end tag
        end_positions = [pos for tag in end_tags if (pos := text.find(tag)) != -1]
        if not end_positions:
            return None
        end = min(end_positions)
        # Trim only the extracted portion; tests expect exact substring
        return text[:end]

    def _parse_tool_call_args(self, text: str) -> str:
        try:
            # Try to parse as JSON
            args = json.loads(text)
        except json.JSONDecodeError:
            try:
                # Try to parse as JSON5
                args = pyjson5.decode(text)
            except pyjson5.Json5DecoderException:
                try:
                    # Try to parse as Python literal
                    args = ast.literal_eval(text)
                except Exception:
                    # Fallback: return the original string
                    args = text
        if not isinstance(args, str):
            # Always convert back to JSON string
            args = json.dumps(args)
        return args

    def _parse_tool_calls(self, text: str) -> List[ToolCall]:
        tool_calls: list[ToolCall] = []
        # Parse globally; wrapper '<|tool_calls|>' may or may not be present.
        section_start = 0
        # section ends at <|calls|> if present, else use end of text
        section_end = text.find("<|calls|>")
        if section_end == -1:
            section_end = len(text)
        i = section_start
        while True:
            begin_tag = "<|tool_call:begin|>"
            name_tag = "<|tool_call:name|>"
            args_tag = "<|tool_call:args|>"
            end_tag = "<|tool_call:end|>"

            b = text.find(begin_tag, i, section_end)
            if b == -1:
                break
            b += len(begin_tag)
            n = text.find(name_tag, b, section_end)
            if n == -1:
                break
            tool_id = text[b:n]
            n += len(name_tag)
            a = text.find(args_tag, n, section_end)
            if a == -1:
                break
            name = text[n:a]
            a += len(args_tag)
            e = text.find(end_tag, a, section_end)
            if e == -1:
                break
            args = text[a:e]
            tool_calls.append(
                ToolCall(
                    id=tool_id,
                    function=FunctionCall(name=name, arguments=self._parse_tool_call_args(args)),
                ))
            i = e + len(end_tag)

        return tool_calls