fix

Author: luukunn

fastdeploy/entrypoints/openai/response_processors.py

@@ -67,13 +67,12 @@ def accumulate_token_ids(self, request_output):
             else:
                 self._multipart_buffer.append({"decode_type": decode_type, "request_output": request_output})
 
-    async def process_response_chat(self, request_outputs, stream, model_status, include_stop_str_in_output):
+    async def process_response_chat(self, request_outputs, stream, include_stop_str_in_output):
         """
         Process a list of responses into a generator that yields each processed response as it's generated.
         Args:
             request_outputs: The list of outputs to be processed.
             stream: Whether or not to stream the output.
-            model_status: Whether or not to show thinking messages.
             include_stop_str_in_output: Whether or not to include stop strings in the output.
         """
         for request_output in request_outputs:
@@ -82,7 +81,6 @@ async def process_response_chat(self, request_outputs, stream, model_status, inc
                 yield self.data_processor.process_response_dict(
                     response_dict=request_output,
                     stream=stream,
-                    model_status=model_status,
                     include_stop_str_in_output=include_stop_str_in_output,
                 )
             elif stream:
@@ -108,7 +106,6 @@ async def process_response_chat(self, request_outputs, stream, model_status, inc
                     self.data_processor.process_response_dict(
                         response_dict=request_output,
                         stream=stream,
-                        model_status=model_status,
                         include_stop_str_in_output=include_stop_str_in_output,
                     )
                     text = {"type": "text", "text": request_output["outputs"]["text"]}
@@ -128,7 +125,6 @@ async def process_response_chat(self, request_outputs, stream, model_status, inc
                             self.data_processor.process_response_dict(
                                 response_dict=part["request_output"],
                                 stream=False,
-                                model_status=model_status,
                                 include_stop_str_in_output=include_stop_str_in_output,
                             )
                             text = {"type": "text", "text": part["request_output"]["outputs"]["text"]}

fastdeploy/input/ernie4_5_processor.py

@@ -240,8 +240,10 @@ def process_request_dict(self, request, max_model_len=None):
         if self.reasoning_parser and self.reasoning_parser.__class__.__name__ == "ErnieX1ReasoningParser":
             request["enable_thinking"] = True
         if self.reasoning_parser:
-            self.model_status_dict["request_id"] = self.reasoning_parser.get_model_status(request["prompt_token_ids"])
-            if self.model_status_dict["request_id"] == "think_start":
+            self.model_status_dict[request["request_id"]] = self.reasoning_parser.get_model_status(
+                request["prompt_token_ids"]
+            )
+            if self.model_status_dict[request["request_id"]] == "think_start":
                 request["enable_thinking"] = True
         data_processor_logger.info(f"Processed request dict: {request}")
         return request
@@ -256,7 +258,6 @@ def process_response(self, response_dict, **kwargs):
         Returns:
             Dict: response contain text fields
         """
-        model_status = kwargs.get("model_status")
         req_id = response_dict.request_id
         token_ids = response_dict.outputs.token_ids
 
@@ -266,7 +267,7 @@ def process_response(self, response_dict, **kwargs):
         full_text = self.tokenizer.decode(token_ids)
         if self.reasoning_parser:
             reasoning_content, text = self.reasoning_parser.extract_reasoning_content(
-                full_text, response_dict, model_status
+                full_text, response_dict, self.model_status_dict[req_id]
             )
             response_dict.outputs.text = text
             response_dict.outputs.reasoning_content = reasoning_content
@@ -310,7 +311,6 @@ def process_response_dict_normal(self, response_dict, **kwargs):
         Returns:
             Dict: response contain text fields
         """
-        enable_thinking = kwargs.get("enable_thinking")
         token_ids = response_dict["outputs"]["token_ids"]
         is_end = response_dict["finished"]
         req_id = response_dict["request_id"]
@@ -321,11 +321,9 @@ def process_response_dict_normal(self, response_dict, **kwargs):
         if is_end:
             full_text = previous_texts + delta_text
             response_dict["outputs"]["text"] = full_text
-            if self.reasoning_parser and (
-                enable_thinking or self.reasoning_parser.__class__.__name__ == "ErnieX1ReasoningParser"
-            ):
+            if self.reasoning_parser:
                 reasoning_content, text = self.reasoning_parser.extract_reasoning_content(
-                    full_text, response_dict, self.model_status_dict.get(req_id)
+                    full_text, response_dict, self.model_status_dict[req_id]
                 )
                 response_dict["outputs"]["text"] = text
                 response_dict["outputs"]["reasoning_content"] = reasoning_content
@@ -352,7 +350,6 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
         Returns:
             Dict: response contain text fields
         """
-        enable_thinking = kwargs.get("enable_thinking")
         is_end = response_dict["finished"]
         req_id = response_dict["request_id"]
         token_ids = response_dict["outputs"]["token_ids"]
@@ -362,17 +359,15 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
                 token_ids = token_ids[:-1]
         delta_text, previous_token_ids, previous_texts = self.ids2tokens(token_ids, req_id)
         response_dict["outputs"]["raw_prediction"] = delta_text
-        if self.reasoning_parser and (
-            enable_thinking or self.reasoning_parser.__class__.__name__ == "ErnieX1ReasoningParser"
-        ):
+        if self.reasoning_parser:
             reasoning_delta_message = self.reasoning_parser.extract_reasoning_content_streaming(
                 previous_texts,
                 previous_texts + delta_text,
                 delta_text,
                 previous_token_ids,
                 previous_token_ids + token_ids,
                 token_ids,
-                self.model_status_dict.get(req_id),
+                self.model_status_dict[req_id],
             )
             response_dict["outputs"]["delta_message"] = reasoning_delta_message
         if self.tool_parser_obj:

fastdeploy/input/ernie4_5_vl_processor/ernie4_5_vl_processor.py

@@ -54,6 +54,7 @@ def __init__(
 
         self.tool_parser_dict = dict()
         self.decode_status = dict()
+        self.model_status_dict = dict()
         self._load_tokenizer()
 
         # Generation config
@@ -255,8 +256,12 @@ def process_request_dict(self, request, max_model_len=None):
             request["max_tokens"] = max(1, max_model_len - len(request["prompt_token_ids"]))
         data_processor_logger.info(f"Processed request {request}")
 
-        if self.reasoning_parser is not None:
-            request["model_status"] = self.reasoning_parser.get_model_status(request["prompt_token_ids"])
+        if self.reasoning_parser:
+            self.model_status_dict[request.request_id] = self.reasoning_parser.get_model_status(
+                request.prompt_token_ids
+            )
+            if self.model_status_dict[request.request_id] == "think_start":
+                request.enable_thinking = True
 
         return request
 
@@ -290,21 +295,3 @@ def pack_outputs(self, outs):
         outs["position_ids"] = np.array(outs["position_ids"], dtype=np.int64)
 
         return outs
-
-    def process_response_dict(self, response_dict, stream, **kwargs):
-        """
-        Preprocess the response
-
-        Args:
-            response_dict (Dict): response for engine, contain ids fields
-
-        Returns:
-            Dict: response contain text fields
-        """
-        enable_thinking = kwargs.pop("enable_thinking", True)
-        if enable_thinking is None:
-            enable_thinking = True
-        if stream:
-            return self.process_response_dict_streaming(response_dict, enable_thinking=enable_thinking, **kwargs)
-        else:
-            return self.process_response_dict_normal(response_dict, enable_thinking=enable_thinking, **kwargs)

fastdeploy/input/text_processor.py

@@ -175,6 +175,7 @@ def __init__(self, model_name_or_path, reasoning_parser_obj=None, tool_parser_ob
             self.generation_config = None
 
         self.decode_status = dict()
+        self.model_status_dict = dict()
         self.tool_parser_dict = dict()
         self.tokenizer = self._load_tokenizer()
         data_processor_logger.info(
@@ -266,8 +267,10 @@ def process_request(self, request, max_model_len=None, **kwargs):
         if request.get("top_p") < _SAMPLING_EPS:
             request.set("top_p", _SAMPLING_EPS)
         if self.reasoning_parser:
-            request.model_status = self.reasoning_parser.get_model_status(request.prompt_token_ids)
-            if request.model_status == "think_start":
+            self.model_status_dict[request.request_id] = self.reasoning_parser.get_model_status(
+                request.prompt_token_ids
+            )
+            if self.model_status_dict[request.request_id] == "think_start":
                 request.enable_thinking = True
 
         data_processor_logger.info(f"Processed request: {request}")
@@ -343,6 +346,12 @@ def process_request_dict(self, request, max_model_len=None, **kwargs):
             request["temperature"] = 1
         if request.get("top_p") < _SAMPLING_EPS:
             request["top_p"] = _SAMPLING_EPS
+        if self.reasoning_parser:
+            self.model_status_dict[request["request_id"]] = self.reasoning_parser.get_model_status(
+                request["prompt_token_ids"]
+            )
+            if self.model_status_dict[request["request_id"]] == "think_start":
+                request["enable_thinking"] = True
 
         data_processor_logger.info(f"Processed request dict: {request}")
         return request
@@ -366,21 +375,22 @@ def process_response(self, response_dict, **kwargs):
         if token_ids[-1] == self.tokenizer.eos_token_id:
             token_ids = token_ids[:-1]
         full_text = self.tokenizer.decode(token_ids)
-
+        response_dict.outputs.text = full_text
         # 模型支持思考,并且支持思考
         if self.reasoning_parser:
-            reasoning_content, text = self.reasoning_parser.extract_reasoning_content(full_text, response_dict)
+            reasoning_content, text = self.reasoning_parser.extract_reasoning_content(
+                full_text, response_dict, self.model_status_dict[req_id]
+            )
             response_dict.outputs.text = text
             response_dict.outputs.reasoning_content = reasoning_content
-        else:
-            # 模型不支持思考,并且没单独设置enable_thinking为false
-            response_dict.outputs.text = full_text
         if self.tool_parser_obj:
             tool_parser = self.tool_parser_obj(self.tokenizer)
             tool_call_info = tool_parser.extract_tool_calls(full_text, response_dict)
             if tool_call_info.tools_called:
                 response_dict.outputs.tool_calls = tool_call_info.tool_calls
                 response_dict.outputs.text = tool_call_info.content
+        if req_id in self.model_status_dict:
+            del self.model_status_dict[req_id]
         data_processor_logger.info(f"req_id:{req_id}, token_ids: {token_ids}")
 
         return response_dict
@@ -395,7 +405,6 @@ def process_response_dict_normal(self, response_dict, **kwargs):
         Returns:
             Dict: response contain text fields
         """
-        enable_thinking = kwargs.get("enable_thinking")
         token_ids = response_dict["outputs"]["token_ids"]
         is_end = response_dict["finished"]
         req_id = response_dict["request_id"]
@@ -406,12 +415,13 @@ def process_response_dict_normal(self, response_dict, **kwargs):
         if is_end:
             full_text = previous_texts + delta_text
             response_dict["outputs"]["raw_prediction"] = full_text
-            if enable_thinking and self.reasoning_parser:
-                reasoning_content, text = self.reasoning_parser.extract_reasoning_content(full_text, response_dict)
+            response_dict["outputs"]["text"] = full_text
+            if self.reasoning_parser:
+                reasoning_content, text = self.reasoning_parser.extract_reasoning_content(
+                    full_text, response_dict, self.model_status_dict[req_id]
+                )
                 response_dict["outputs"]["text"] = text
                 response_dict["outputs"]["reasoning_content"] = reasoning_content
-            else:
-                response_dict["outputs"]["text"] = full_text
             if self.tool_parser_obj:
                 tool_parser = self.tool_parser_obj(self.tokenizer)
                 tool_call_info = tool_parser.extract_tool_calls(full_text, response_dict)
@@ -432,7 +442,6 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
         Returns:
             Dict: response contain text fields
         """
-        enable_thinking = kwargs.get("enable_thinking")
         is_end = response_dict["finished"]
         req_id = response_dict["request_id"]
         token_ids = response_dict["outputs"]["token_ids"]
@@ -442,16 +451,15 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
                 token_ids = token_ids[:-1]
         delta_text, previous_token_ids, previous_texts = self.ids2tokens(token_ids, req_id)
         response_dict["outputs"]["raw_prediction"] = delta_text
-        if self.reasoning_parser and (
-            enable_thinking or self.reasoning_parser.__class__.__name__ == "ErnieX1ReasoningParser"
-        ):
+        if self.reasoning_parser:
             reasoning_delta_message = self.reasoning_parser.extract_reasoning_content_streaming(
                 previous_texts,
                 previous_texts + delta_text,
                 delta_text,
                 previous_token_ids,
                 previous_token_ids + token_ids,
                 token_ids,
+                self.model_status_dict[req_id],
             )
             response_dict["outputs"]["delta_message"] = reasoning_delta_message
         if self.tool_parser_obj:
@@ -475,6 +483,8 @@ def process_response_dict_streaming(self, response_dict, **kwargs):
             del self.decode_status[req_id]
             if req_id in self.tool_parser_dict:
                 del self.tool_parser_dict[req_id]
+            if req_id in self.model_status_dict:
+                del self.model_status_dict[req_id]
         return response_dict
 
     def process_response_dict(self, response_dict, **kwargs):
@@ -487,16 +497,12 @@ def process_response_dict(self, response_dict, **kwargs):
         Returns:
             Dict: response contain text fields
         """
-        enable_thinking = kwargs.pop("enable_thinking", True)
-        if enable_thinking is None:
-            enable_thinking = True
         stream = kwargs.get("stream", True)
         if stream:
-            return self.process_response_dict_streaming(response_dict, enable_thinking=enable_thinking, **kwargs)
+            return self.process_response_dict_streaming(response_dict, **kwargs)
         else:
             return self.process_response_dict_normal(
                 response_dict=response_dict,
-                enable_thinking=enable_thinking,
                 **kwargs,
             )