Skip to content

vllm.entrypoints.openai.tool_parsers.phi4mini_tool_parser

logger module-attribute 

logger = init_logger(__name__)

Phi4MiniJsonToolParser

Bases: ToolParser

Tool call parser for phi-4-mini models intended for use with the examples/tool_chat_template_llama.jinja template.

Used when --enable-auto-tool-choice --tool-call-parser phi4_mini_json
are all set

Source code in vllm/entrypoints/openai/tool_parsers/phi4mini_tool_parser.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@ToolParserManager.register_module("phi4_mini_json")
class Phi4MiniJsonToolParser(ToolParser):
    """
    Tool call parser for phi-4-mini models intended for use with the
    examples/tool_chat_template_llama.jinja template.

    Used when --enable-auto-tool-choice --tool-call-parser phi4_mini_json  
    are all set
    """

    def __init__(self, tokenizer: PreTrainedTokenizerBase) -> None:
        super().__init__(tokenizer)

        # initialize properties used for state when parsing tool calls in
        # streaming mode
        self.prev_tool_call_arr: list[dict[str, Any]] = []
        self.current_tool_id: int = -1
        self.current_tool_name_sent: bool = False
        self.streamed_args_for_tool: list[str] = [
        ]  # map what has been streamed for each tool so far to a list
        self.bot_token: str = "functools"

    def extract_tool_calls(
            self, model_output: str,
            request: ChatCompletionRequest) -> ExtractedToolCallInformation:
        """
        Extract the tool calls from a complete model response.
        """
        logger.debug("Model output: %s", model_output)

        pattern = r'functools\[(.*?)\]'
        matches = re.search(pattern, model_output, re.DOTALL)

        if not matches:
            logger.debug("No function calls found")
            return ExtractedToolCallInformation(tools_called=False,
                                                tool_calls=[],
                                                content=model_output)

        try:
            function_call_arr: list[dict[str, Any]] = []
            try:
                json_content = '[' + matches.group(1) + ']'

                function_call_arr = json.loads(json_content)
                logger.debug("Successfully extracted %d function calls",
                             len(function_call_arr))
            except json.JSONDecodeError as e:
                logger.error(
                    "Failed to parse function calls from model output. "
                    "Error: %s", str(e))

            tool_calls: list[ToolCall] = [
                ToolCall(
                    id=random_tool_call_id(),
                    type="function",
                    function=FunctionCall(
                        name=raw_function_call["name"],
                        # function call args are JSON but as a string
                        arguments=json.dumps(
                            raw_function_call["arguments"]
                            if "arguments" in raw_function_call else
                            raw_function_call["parameters"],
                            ensure_ascii=False),
                    )) for raw_function_call in function_call_arr
            ]

            # get any content before the tool call
            ret = ExtractedToolCallInformation(tools_called=True,
                                               tool_calls=tool_calls,
                                               content=None)
            return ret

        except Exception:
            return ExtractedToolCallInformation(tools_called=False,
                                                tool_calls=[],
                                                content=model_output)

    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,
    ) -> Optional[DeltaMessage]:

        return None

bot_token instance-attribute 

bot_token: str = 'functools'

current_tool_id instance-attribute 

current_tool_id: int = -1

current_tool_name_sent instance-attribute 

current_tool_name_sent: bool = False

prev_tool_call_arr instance-attribute 

prev_tool_call_arr: list[dict[str, Any]] = []

streamed_args_for_tool instance-attribute 

streamed_args_for_tool: list[str] = []

__init__ 

__init__(tokenizer: PreTrainedTokenizerBase) -> None
Source code in vllm/entrypoints/openai/tool_parsers/phi4mini_tool_parser.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, tokenizer: PreTrainedTokenizerBase) -> None:
    super().__init__(tokenizer)

    # initialize properties used for state when parsing tool calls in
    # streaming mode
    self.prev_tool_call_arr: list[dict[str, Any]] = []
    self.current_tool_id: int = -1
    self.current_tool_name_sent: bool = False
    self.streamed_args_for_tool: list[str] = [
    ]  # map what has been streamed for each tool so far to a list
    self.bot_token: str = "functools"

extract_tool_calls 

extract_tool_calls(
    model_output: str, request: ChatCompletionRequest
) -> ExtractedToolCallInformation

Extract the tool calls from a complete model response.

Source code in vllm/entrypoints/openai/tool_parsers/phi4mini_tool_parser.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def extract_tool_calls(
        self, model_output: str,
        request: ChatCompletionRequest) -> ExtractedToolCallInformation:
    """
    Extract the tool calls from a complete model response.
    """
    logger.debug("Model output: %s", model_output)

    pattern = r'functools\[(.*?)\]'
    matches = re.search(pattern, model_output, re.DOTALL)

    if not matches:
        logger.debug("No function calls found")
        return ExtractedToolCallInformation(tools_called=False,
                                            tool_calls=[],
                                            content=model_output)

    try:
        function_call_arr: list[dict[str, Any]] = []
        try:
            json_content = '[' + matches.group(1) + ']'

            function_call_arr = json.loads(json_content)
            logger.debug("Successfully extracted %d function calls",
                         len(function_call_arr))
        except json.JSONDecodeError as e:
            logger.error(
                "Failed to parse function calls from model output. "
                "Error: %s", str(e))

        tool_calls: list[ToolCall] = [
            ToolCall(
                id=random_tool_call_id(),
                type="function",
                function=FunctionCall(
                    name=raw_function_call["name"],
                    # function call args are JSON but as a string
                    arguments=json.dumps(
                        raw_function_call["arguments"]
                        if "arguments" in raw_function_call else
                        raw_function_call["parameters"],
                        ensure_ascii=False),
                )) for raw_function_call in function_call_arr
        ]

        # get any content before the tool call
        ret = ExtractedToolCallInformation(tools_called=True,
                                           tool_calls=tool_calls,
                                           content=None)
        return ret

    except Exception:
        return ExtractedToolCallInformation(tools_called=False,
                                            tool_calls=[],
                                            content=model_output)

extract_tool_calls_streaming 

extract_tool_calls_streaming(
    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,
) -> Optional[DeltaMessage]
Source code in vllm/entrypoints/openai/tool_parsers/phi4mini_tool_parser.py 
101
102
103
104
105
106
107
108
109
110
111
112
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,
) -> Optional[DeltaMessage]:

    return None