mm support structured output

Author: kevincheng2

fastdeploy/config.py

@@ -144,6 +144,7 @@ class MoEConfig:
     im_patch_id = (
         100295  # multimodality, TODO(liuyuanle): read from config.json
     )
+    reasoning_parser: Optional[str] = None
 
 
 @dataclass

fastdeploy/engine/config.py

@@ -656,7 +656,8 @@ def postprocess(self):
             self.max_model_len // self.cache_config.block_size)
 
         if self.guided_decoding_backend == "auto":
-            if self.enable_mm:
+            if current_platform.is_xpu() or self.speculative_config.method is not None:
+                llm_logger.warning("Speculative Decoding and XPU currently do not support Guided decoding, set off.")
                 self.guided_decoding_backend = "off"
             else:
                 self.guided_decoding_backend = "xgrammar"
@@ -718,10 +719,9 @@ def check(self):
                 f"Only support xgrammar、auto guided decoding backend, but got {self.guided_decoding_backend}."
 
             if self.guided_decoding_backend != "off":
-                # TODO: mm support guided_decoding
-                assert self.enable_mm is False, "Multimodal model currently do not support guided_decoding"
-
                 # TODO: speculative decoding support guided_decoding
+                assert self.speculative_config.method is None, \
+                "speculative decoding currently do not support guided_decoding"
 
                 # TODO: xpu support guided_decoding
                 assert not current_platform.is_xpu(
@@ -749,7 +749,7 @@ def print(self, file=None):
             if k == "generation_config" and v is not None:
                 for gck, gcv in v.to_dict().items():
                     llm_logger.info("{:<20}:{:<6}{}".format(gck, "", gcv))
-            elif k == "cache_config" or k == "model_config" or k == "scheduler_config" or k == "parallel_config":
+            elif k in ["cache_config", "model_config", "scheduler_config", "scheduler_config", "parallel_config", "speculative_config"]:
                 v.print()
             else:
                 llm_logger.info("{:<20}:{:<6}{}".format(k, "", v))

fastdeploy/engine/engine.py

@@ -385,6 +385,13 @@ def _insert_zmq_task_to_scheduler(self):
                     llm_logger.debug(f"Receive request: {request}")
 
                     err_msg = None
+                    if ((request.guided_json is not None
+                    or request.guided_regex is not None
+                    or request.structural_tag is not None
+                    or request.guided_grammar is not None) and self.guided_decoding_checker is None):
+                        err_msg = "guided_backend is None, use --guided-decoding-backend to " \
+                                  "specify the backend at server startup."
+
                     if self.guided_decoding_checker is not None:
                         request, err_msg = self.guided_decoding_checker.schema_format(
                             request)
@@ -473,6 +480,14 @@ def add_requests(self, task, sampling_params=None, **kwargs):
             llm_logger.error(error_msg)
             raise EngineError(error_msg, error_code=400)
 
+        if ((request.guided_json is not None
+        or request.guided_regex is not None
+        or request.structural_tag is not None
+        or request.guided_grammar is not None) and self.guided_decoding_checker is None):
+            err_msg = "guided_backend is None, use --guided-decoding-backend to specify the backend at server startup."
+            llm_logger.error(err_msg)
+            raise EngineError(err_msg, error_code=400)
+
         if self.guided_decoding_checker is not None:
             request, err_msg = self.guided_decoding_checker.schema_format(
                 request)
@@ -1021,8 +1036,8 @@ def _start_worker_service(self):
         py_script = os.path.join(current_dir_path, worker_path)
 
         ori_vocab_size = (
-            len(self.data_processor.tokenizer.sp_model) 
-            if hasattr(self.data_processor.tokenizer, 'sp_model') 
+            len(self.data_processor.tokenizer.sp_model)
+            if hasattr(self.data_processor.tokenizer, 'sp_model')
             else len(self.data_processor.tokenizer.vocab)
         )
 
@@ -1053,7 +1068,8 @@ def _start_worker_service(self):
             f" --speculative_model_quantization {self.cfg.speculative_config.quantization}"
             f" --max_capture_batch_size {self.cfg.max_capture_batch_size}"
             f" --guided_decoding_backend {self.cfg.guided_decoding_backend}"
-            f" --load_strategy {self.cfg.model_config.load_strategy}")
+            f" --load_strategy {self.cfg.model_config.load_strategy}"
+            f" --reasoning_parser {self.cfg.reasoning_parser}")
 
         worker_append_flag = {
             "enable_expert_parallel":

fastdeploy/engine/sampling_params.py

@@ -90,6 +90,7 @@ class SamplingParams:
     min_tokens: int = 1
     logprobs: Optional[int] = None
     bad_words: Optional[List[str]] = None
+    guided_decoding: Optional[GuidedDecodingParams] = None
 
     @classmethod
     def from_dict(cls, req_dict: dict[str, Any]) -> "SamplingParams":
@@ -118,7 +119,8 @@ def from_optional(cls,
                       reasoning_max_tokens=None,
                       min_tokens=1,
                       logprobs=None,
-                      bad_words=None) -> "SamplingParams":
+                      bad_words=None,
+                      guided_decoding=None) -> "SamplingParams":
         """Create instance from command line arguments"""
         return cls(n=1 if n is None else n,
                    best_of=best_of,
@@ -137,7 +139,8 @@ def from_optional(cls,
                    reasoning_max_tokens=reasoning_max_tokens,
                    min_tokens=min_tokens,
                    logprobs=logprobs,
-                   bad_words=bad_words)
+                   bad_words=bad_words,
+                   guided_decoding=guided_decoding)
 
     def __post_init__(self):
         if self.seed is None:
@@ -193,6 +196,9 @@ def _verify_args(self) -> None:
             raise ValueError("seed must be in [0, 922337203685477580], got "
                              f"{self.seed}.")
 
+        if self.guided_decoding is not None:
+            self.guided_decoding._verify_args()
+
     def update_from_tokenizer(self, tokenizer):
         """
         # TODO: Implement stop tokens and bad words support
@@ -210,3 +216,45 @@ class BeamSearchParams:
     temperature: float = 0.0
     length_penalty: float = 1.0
     include_stop_str_in_output: bool = False
+
+
+@dataclass
+class GuidedDecodingParams:
+    """Guided decoding parameters for text generation."""
+    json: Optional[Union[str, dict]] = None
+    regex: Optional[str] = None
+    choice: Optional[List[str]] = None
+    grammar: Optional[str] = None
+    json_object: Optional[bool] = None
+    structural_tag: Optional[str] = None
+
+    def to_dict(self):
+        """convert to dict"""
+        key_dict = {
+            "guided_json": self.json,
+            "guided_regex": self.regex,
+            "guided_choice": self.choice,
+            "guided_grammar": self.grammar,
+            "structural_tag": self.structural_tag,
+            "guided_json_object": self.json_object,
+        }
+
+        guided_dict = {}
+        for key, value in key_dict.items():
+            if value is not None:
+                guided_dict[key] = value
+        return guided_dict
+
+    def _verify_args(self):
+        """Verify the arguments."""
+        guided_count = sum([
+            self.json is not None, self.regex is not None, self.choice
+            is not None, self.grammar is not None, self.json_object
+            is not None, self.structural_tag is not None
+        ])
+
+        if guided_count > 1:
+            raise ValueError(
+                "You can only use one kind of guided decoding "
+                "('json', 'json_object', 'regex', 'choice', 'grammar', 'structural_tag')."
+            )

fastdeploy/entrypoints/llm.py

@@ -258,6 +258,9 @@ def _add_request(
             if chat_template_kwargs is not None:
                 enable_thinking = chat_template_kwargs.get(
                     "enable_thinking", None)
+            if current_sampling_params.guided_decoding is not None:
+                guided_decoding_dict = current_sampling_params.guided_decoding.to_dict()
+                tasks.update(guided_decoding_dict)
             self.llm_engine.add_requests(tasks,
                                          current_sampling_params,
                                          enable_thinking=enable_thinking)

fastdeploy/model_executor/guided_decoding/__init__.py

@@ -15,8 +15,10 @@
 """
 
 # from fastdeploy.config import FDConfig
+from fastdeploy.model_executor.guided_decoding.base_guided_decoding import (
+    BackendBase, BaseChecker, LogitsProcessorBase)
 
-__all__ = ['get_guided_backend', 'schema_checker']
+__all__ = ['get_guided_backend', 'schema_checker', 'LogitsProcessorBase', 'BackendBase', 'BaseChecker']
 
 
 def get_guided_backend(

fastdeploy/model_executor/guided_decoding/base_guided_decoding.py

@@ -19,6 +19,7 @@
 
 from fastdeploy.config import FDConfig
 from fastdeploy.engine.request import Request
+from fastdeploy.reasoning import ReasoningParserManager
 from fastdeploy.utils import llm_logger
 
 
@@ -34,8 +35,9 @@ class LogitsProcessorBase:
         None (all state should be managed by subclasses)
     """
 
-    def __init__(self):
-        pass
+    def __init__(self, enable_reasoning):
+        self.reasoning_ended = False
+        self.enable_reasoning = enable_reasoning
 
     def fill_token_bitmask(self, token_bitmask, idx):
         """
@@ -136,8 +138,13 @@ def __init__(self, fd_config: FDConfig):
         self.fd_config = fd_config
         self.executor = ThreadPoolExecutor()
         self.max_cache_size = 2048
+        self.reasoning_parser = None
 
         self.hf_tokenizer = self._get_tokenizer_hf()
+        if self.fd_config.model_config.reasoning_parser:
+            reasoning_parser_obj = ReasoningParserManager.get_reasoning_parser(
+                self.fd_config.model_config.reasoning_parser)
+            self.reasoning_parser = reasoning_parser_obj(self.hf_tokenizer)
 
     def _create_processor(self):
         """
@@ -148,71 +155,89 @@ def _create_processor(self):
         """
         raise NotImplementedError()
 
-    def _json_processor(self, schemata):
+    def _json_processor(self, schemata, enable_thinking=False):
         """
         Process JSON schemata.
 
         Args:
             schemata (str): The schemata string.
+            enable_thinking (bool): Whether to enable thinking mode.
 
         Raises:
             NotImplementedError: This method should be implemented in subclasses.
         """
         raise NotImplementedError()
 
-    def _regex_processor(self, schemata):
+    def _regex_processor(self, schemata, enable_thinking=False):
         """
         Process regular expression schemata.
 
         Args:
             schemata (str): The schemata string.
+            enable_thinking (bool): Whether to enable thinking mode.
 
         Raises:
             NotImplementedError: This method should be implemented in subclasses.
         """
         raise NotImplementedError()
 
-    def _grammar_processor(self, schemata):
+    def _grammar_processor(self, schemata, enable_thinking=False):
         """
         Process grammar schemata.
 
         Args:
             schemata (str): The schemata string.
+            enable_thinking (bool): Whether to enable thinking mode.
 
         Raises:
             NotImplementedError: This method should be implemented in subclasses.
         """
         raise NotImplementedError()
 
-    def _structural_tag_processor(self, schemata):
+    def _structural_tag_processor(self, schemata, enable_thinking=False):
         """
         Process structural tag schemata.
 
         Args:
             schemata (str): The schemata string.
+            enable_thinking (bool): Whether to enable thinking mode.
 
         Raises:
             NotImplementedError: This method should be implemented in subclasses.
         """
         raise NotImplementedError()
 
-    def _unsupported_processor_type(self, key_type, schemata):
+    def _unsupported_processor_type(self, key_type, schemata, enable_thinking=False):
         """
         Process unsupported type.
 
         Args:
             key_type (str): The key type string.
             schemata (str): The schemata string.
+            enable_thinking (bool): Whether to enable thinking mode.
         """
         raise Exception(f"Unsupported processor type {key_type}.")
 
+    def get_reasoning_parser(self):
+        """
+        Get reasoning parser object.
+
+        Returns:
+            ReasoningParser: Reasoning parser object or None
+        """
+        return self.reasoning_parser
+
     def _init_logits_processor(
-            self, schemata_key: tuple[str, str]) -> LogitsProcessorBase:
+            self,
+            schemata_key: tuple[str, str],
+            enable_thinking: bool = False,
+        ) -> LogitsProcessorBase:
         """
         init logits processor by type and schemata.
 
         Args:
             schemata_key (tuple[str, str]): Tuple containing processor type and schema string
+            enable_thinking (bool): Whether to enable thinking step
 
         Returns:
             LogitsProcessorBase: Initialized logits processor instance
@@ -222,20 +247,21 @@ def _init_logits_processor(
         """
         key_type, schemata = schemata_key
         if key_type == "json":
-            return self._json_processor(schemata)
+            return self._json_processor(schemata, enable_thinking)
         elif key_type == "regex":
-            return self._regex_processor(schemata)
+            return self._regex_processor(schemata, enable_thinking)
         elif key_type == "grammar":
-            return self._grammar_processor(schemata)
+            return self._grammar_processor(schemata, enable_thinking)
         elif key_type == "structural_tag":
-            return self._structural_tag_processor(schemata)
+            return self._structural_tag_processor(schemata, enable_thinking)
         else:
             llm_logger.error(f"Unsupported processor type {key_type}.")
             return None
 
     def get_logits_processor(
             self,
-            schemata_key: tuple[str, str]) -> tuple[LogitsProcessorBase, bool]:
+            schemata_key: tuple[str, str],
+            enable_thinking: bool = False) -> tuple[LogitsProcessorBase, bool]:
         """
         get logits processor by key from cache or create new one.
 
@@ -249,8 +275,10 @@ def get_logits_processor(
         """
         value = self.cache.get(schemata_key, None)
         if value:
-            return value.copy(), True
-        value = self.executor.submit(self._init_logits_processor, schemata_key)
+            value_copy = value.copy()
+            value_copy.enable_reasoning = enable_thinking
+            return value_copy, True
+        value = self.executor.submit(self._init_logits_processor, schemata_key, enable_thinking)
         return value, False
 
     def _get_tokenizer_hf(self):
@@ -269,7 +297,8 @@ def _get_tokenizer_hf(self):
         try:
             architectures = self.fd_config.model_config.architectures
             if "Ernie4_5_MoeForCausalLM" not in architectures \
-                and "Ernie4_5_ForCausalLM" not in architectures:
+            and "Ernie4_5_ForCausalLM" not in architectures \
+            and "Ernie4_5_VLMoeForConditionalGeneration" not in architectures:
 
                 from transformers import AutoTokenizer, PreTrainedTokenizerFast
                 tokenizer = AutoTokenizer.from_pretrained(