add tool parser

Author: luukunn

fastdeploy/engine/args_utils.py

@@ -15,10 +15,10 @@
 """
 
 import json
+import os
 from dataclasses import asdict, dataclass
 from dataclasses import fields as dataclass_fields
 from typing import Any, Dict, List, Optional
-import os
 
 from fastdeploy.config import (
     CacheConfig,
@@ -93,6 +93,14 @@ class EngineArgs:
     """
     specifies the reasoning parser to use for extracting reasoning content from the model output
     """
+    tool_call_parser: str = None
+    """
+    specifies the tool call parser  to use for extracting tool call from the model output
+    """
+    tool_parser_plugin: str = None
+    """
+    tool parser plugin used to register user defined tool parsers
+    """
     enable_mm: bool = False
     """
     Flags to enable multi-modal model
@@ -421,6 +429,18 @@ def add_cli_args(parser: FlexibleArgumentParser) -> FlexibleArgumentParser:
             help="Flag specifies the reasoning parser to use for extracting "
             "reasoning content from the model output",
         )
+        model_group.add_argument(
+            "--tool-call-parser",
+            type=str,
+            default=EngineArgs.tool_call_parser,
+            help="Flag specifies the tool call parser to use for extracting" "tool call from the model output",
+        )
+        model_group.add_argument(
+            "--tool-parser-plugin",
+            type=str,
+            default=EngineArgs.tool_parser_plugin,
+            help="tool parser plugin used to register user defined tool parsers",
+        )
         model_group.add_argument(
             "--speculative-config",
             type=json.loads,
@@ -866,10 +886,10 @@ def create_engine_config(self) -> Config:
             if self.enable_chunked_prefill:
                 self.max_num_batched_tokens = 2048
             else:
-                if not int(os.getenv('ENABLE_V1_KVCACHE_SCHEDULER', '0')):
+                if not int(os.getenv("ENABLE_V1_KVCACHE_SCHEDULER", "0")):
                     self.max_num_batched_tokens = self.max_model_len
                 else:
-                    self.max_num_batched_tokens = 8192
+                    self.max_num_batched_tokens = 8192  # if set to max_model_len, it's easy to be OOM
 
         all_dict = asdict(self)
         all_dict["model_cfg"] = model_cfg
@@ -908,6 +928,7 @@ def create_engine_config(self) -> Config:
             mm_processor_kwargs=self.mm_processor_kwargs,
             enable_mm=self.enable_mm,
             reasoning_parser=self.reasoning_parser,
+            tool_parser=self.tool_call_parser,
             splitwise_role=self.splitwise_role,
             innode_prefill_ports=self.innode_prefill_ports,
             max_num_partial_prefills=self.max_num_partial_prefills,

fastdeploy/engine/config.py

@@ -85,6 +85,7 @@ def __init__(
         max_long_partial_prefills: int = 1,
         long_prefill_token_threshold: int = 0,
         reasoning_parser: str = None,
+        tool_parser: str = None,
         guided_decoding_backend: Optional[str] = None,
         disable_any_whitespace: bool = False,
         enable_logprob: bool = False,
@@ -165,6 +166,7 @@ def __init__(
         self.max_long_partial_prefills = max_long_partial_prefills
         self.long_prefill_token_threshold = long_prefill_token_threshold
         self.reasoning_parser = reasoning_parser
+        self.tool_parser = tool_parser
         self.graph_optimization_config = graph_optimization_config
         self.early_stop_config = early_stop_config
         self.guided_decoding_backend = guided_decoding_backend
@@ -236,10 +238,10 @@ def postprocess(self):
             if self.cache_config.enable_chunked_prefill:
                 self.max_num_batched_tokens = 2048
             else:
-                if not int(os.getenv('ENABLE_V1_KVCACHE_SCHEDULER', '0')):
+                if not int(os.getenv("ENABLE_V1_KVCACHE_SCHEDULER", "0")):
                     self.max_num_batched_tokens = self.max_model_len
                 else:
-                    self.max_num_batched_tokens = 8192
+                    self.max_num_batched_tokens = 8192  # if set to max_model_len, it's easy to be OOM
 
         if self.long_prefill_token_threshold == 0:
             self.long_prefill_token_threshold = int(self.max_model_len * 0.04)
@@ -287,7 +289,7 @@ def check(self):
         )
 
         if not self.cache_config.enable_chunked_prefill:
-            if not int(os.getenv('ENABLE_V1_KVCACHE_SCHEDULER', '0')):
+            if not int(os.getenv("ENABLE_V1_KVCACHE_SCHEDULER", "0")):
                 assert self.max_num_batched_tokens >= self.max_model_len, (
                     f"max_num_batched_tokens: {self.max_num_batched_tokens} "
                     f"should be larger than or equal to max_model_len: {self.max_model_len}"

fastdeploy/engine/engine.py

@@ -106,6 +106,7 @@ def __init__(self, cfg):
             cfg.limit_mm_per_prompt,
             cfg.mm_processor_kwargs,
             cfg.enable_mm,
+            cfg.tool_parser,
         )
 
         self.start_queue_service()

fastdeploy/engine/request.py

@@ -24,6 +24,7 @@
 import numpy as np
 
 from fastdeploy.engine.sampling_params import SamplingParams
+from fastdeploy.entrypoints.openai.protocol import ToolCall
 from fastdeploy.utils import data_processor_logger
 from fastdeploy.worker.output import LogprobsLists, SampleLogprobs
 
@@ -249,6 +250,7 @@ class CompletionOutput:
     draft_token_ids: list[int] = None
     text: Optional[str] = None
     reasoning_content: Optional[str] = None
+    tool_calls: Optional[ToolCall] = None
 
     def to_dict(self):
         """

fastdeploy/entrypoints/chat_utils.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 """
 
+import uuid
 from copy import deepcopy
 from typing import List, Literal, Union
 from urllib.parse import urlparse
@@ -156,3 +157,7 @@ def parse_chat_messages(messages):
 
         conversation.append({"role": role, "content": parsed_content})
     return conversation
+
+
+def random_tool_call_id() -> str:
+    return f"chatcmpl-tool-{str(uuid.uuid4().hex)}"

fastdeploy/entrypoints/engine_client.py

@@ -45,13 +45,15 @@ def __init__(
         data_parallel_size=1,
         enable_logprob=False,
         workers=1,
+        tool_parser=None,
     ):
         input_processor = InputPreprocessor(
             tokenizer,
             reasoning_parser,
             limit_mm_per_prompt,
             mm_processor_kwargs,
             enable_mm,
+            tool_parser,
         )
         self.enable_logprob = enable_logprob
         self.enable_mm = enable_mm

fastdeploy/entrypoints/llm.py

@@ -28,8 +28,7 @@
 from fastdeploy.engine.args_utils import EngineArgs
 from fastdeploy.engine.engine import LLMEngine
 from fastdeploy.engine.sampling_params import SamplingParams
-
-# from fastdeploy.entrypoints.chat_utils import ChatCompletionMessageParam
+from fastdeploy.entrypoints.openai.tool_parsers import ToolParserManager
 from fastdeploy.utils import llm_logger, retrive_model_from_server
 from fastdeploy.worker.output import Logprob, LogprobsLists
 
@@ -73,6 +72,9 @@ def __init__(
         **kwargs,
     ):
         model = retrive_model_from_server(model, revision)
+        tool_parser_plugin = kwargs.get("tool_parser_plugin")
+        if tool_parser_plugin:
+            ToolParserManager.import_tool_parser(tool_parser_plugin)
         engine_args = EngineArgs(
             model=model,
             tokenizer=tokenizer,

fastdeploy/entrypoints/openai/api_server.py

@@ -41,6 +41,7 @@
 )
 from fastdeploy.entrypoints.openai.serving_chat import OpenAIServingChat
 from fastdeploy.entrypoints.openai.serving_completion import OpenAIServingCompletion
+from fastdeploy.entrypoints.openai.tool_parsers import ToolParserManager
 from fastdeploy.metrics.metrics import (
     EXCLUDE_LABELS,
     cleanup_prometheus_files,
@@ -73,7 +74,8 @@
 parser = EngineArgs.add_cli_args(parser)
 args = parser.parse_args()
 args.model = retrive_model_from_server(args.model, args.revision)
-
+if args.tool_parser_plugin:
+    ToolParserManager.import_tool_parser(args.tool_parser_plugin)
 llm_engine = None
 
 
@@ -126,6 +128,7 @@ async def lifespan(app: FastAPI):
         args.data_parallel_size,
         args.enable_logprob,
         args.workers,
+        args.tool_call_parser,
     )
     app.state.dynamic_load_weight = args.dynamic_load_weight
     chat_handler = OpenAIServingChat(engine_client, pid, args.ips, args.max_waiting_time)

fastdeploy/entrypoints/openai/protocol.py

@@ -72,7 +72,6 @@ class ToolCall(BaseModel):
     id: str = None
     type: Literal["function"] = "function"
     function: FunctionCall
-    index: int
 
 
 class DeltaFunctionCall(BaseModel):
@@ -96,6 +95,18 @@ class DeltaToolCall(BaseModel):
     function: Optional[DeltaFunctionCall] = None
 
 
+class ExtractedToolCallInformation(BaseModel):
+    # indicate if tools were called
+    tools_called: bool
+
+    # extracted tool calls
+    tool_calls: Optional[list[ToolCall]] = None
+
+    # content - per OpenAI spec, content AND tool calls can be returned rarely
+    # But some models will do this intentionally
+    content: Optional[str] = None
+
+
 class FunctionDefinition(BaseModel):
     """
     Function definition.

fastdeploy/entrypoints/openai/serving_chat.py

@@ -141,6 +141,7 @@ async def chat_completion_stream_generator(
         previous_num_tokens = 0
         num_prompt_tokens = 0
         num_choices = 1
+        tool_called = False
         max_streaming_response_tokens = (
             request.max_streaming_response_tokens
             if request.max_streaming_response_tokens is not None
@@ -244,20 +245,28 @@ async def chat_completion_stream_generator(
                     output = res["outputs"]
                     delta_text = output["text"]
                     output_top_logprobs = output["top_logprobs"]
+                    previous_num_tokens += len(output["token_ids"])
                     logprobs_res: Optional[LogProbs] = None
                     if request.logprobs and output_top_logprobs is not None:
                         logprobs_res = self._create_chat_logprobs(
                             output_top_logprobs, request.logprobs, request.top_logprobs
                         )
-
-                    previous_num_tokens += len(output["token_ids"])
-                    delta_message = DeltaMessage(
-                        content=delta_text,
-                        reasoning_content=output.get("reasoning_content"),
-                        prompt_token_ids=None,
-                        completion_token_ids=None,
-                        tool_calls=output.get("tool_call_content", []),
-                    )
+                    if self.engine_client.data_processor.tool_parser_obj and not res["finished"]:
+                        tool_delta_message = output["tool_delta_message"]
+                        if tool_delta_message is None:
+                            continue
+                        delta_message = tool_delta_message
+                        delta_message.reasoning_content = output.get("reasoning_content")
+                        if delta_message.tool_calls:
+                            tool_called = True
+                    else:
+                        delta_message = DeltaMessage(
+                            content=delta_text,
+                            reasoning_content=output.get("reasoning_content"),
+                            prompt_token_ids=None,
+                            completion_token_ids=None,
+                            tool_calls=None,
+                        )
 
                     choice = ChatCompletionResponseStreamChoice(
                         index=0,
@@ -274,10 +283,7 @@ async def chat_completion_stream_generator(
                         max_tokens = request.max_completion_tokens or request.max_tokens
                         if has_no_token_limit or previous_num_tokens != max_tokens:
                             choice.finish_reason = "stop"
-                            if (
-                                self.engine_client.reasoning_parser == "ernie_x1"
-                                and output.get("finish_reason", "") == "tool_calls"
-                            ):
+                            if tool_called:
                                 choice.finish_reason = "tool_calls"
                         else:
                             choice.finish_reason = "length"
@@ -414,7 +420,7 @@ async def chat_completion_full_generator(
             role="assistant",
             content=output["text"],
             reasoning_content=output.get("reasoning_content"),
-            tool_calls=output.get("tool_call_content"),
+            tool_calls=output.get("tool_call"),
             prompt_token_ids=prompt_token_ids if request.return_token_ids else None,
             completion_token_ids=completion_token_ids if request.return_token_ids else None,
             text_after_process=text_after_process if request.return_token_ids else None,
@@ -434,7 +440,7 @@ async def chat_completion_full_generator(
         max_tokens = request.max_completion_tokens or request.max_tokens
         if has_no_token_limit or previous_num_tokens != max_tokens:
             choice.finish_reason = "stop"
-            if self.engine_client.reasoning_parser == "ernie_x1" and output.get("finish_reason", "") == "tool_calls":
+            if output.get("tool_call"):
                 choice.finish_reason = "tool_calls"
         else:
             choice.finish_reason = "length"