[DataProcessor]Remove ENABLE_V1_DATA_PROCESSOR (#7052)

* remove ENABLE_V1_DATA_PROCESSOR

* fix unit test

* fix unit test

Author: luukunn

fastdeploy/engine/async_llm.py

@@ -446,7 +446,7 @@ async def add_request(
                 )
             if envs.ZMQ_SEND_BATCH_DATA and self.connection_manager is not None:
                 request["zmq_worker_pid"] = self.connection_manager.worker_pid
-            if not envs.ENABLE_V1_DATA_PROCESSOR and self.cfg.model_config.enable_mm:
+            if self.cfg.model_config.enable_mm:
                 self.request_client.send_pyobj(request)
             else:
                 self.request_client.send_json(request)
@@ -543,8 +543,7 @@ async def generate(
                             )
                         else:
                             processed_output = response_item
-                        if not envs.ENABLE_V1_DATA_PROCESSOR:
-                            processed_output = RequestOutput.from_dict(processed_output)
+                        processed_output = RequestOutput.from_dict(processed_output)
                         # Enrich outputs with prompt metadata on the first packet
                         if req_id:
                             prompt_meta = self._prompt_metadata.get(req_id)

fastdeploy/engine/common_engine.py

@@ -1186,7 +1186,7 @@ def _insert_zmq_task_to_scheduler(self):
         while self.running:
             try:
                 block = True if len(added_requests) == 0 else False
-                if not self.cfg.model_config.enable_mm and not envs.ENABLE_V1_DATA_PROCESSOR:
+                if not self.cfg.model_config.enable_mm:
                     err, data = self.recv_request_server.receive_json_once(block)
                 else:
                     err, data = self.recv_request_server.receive_pyobj_once(block)
@@ -1243,8 +1243,7 @@ def _insert_zmq_task_to_scheduler(self):
                         continue
                     err_msg = None
                     try:
-                        if not envs.ENABLE_V1_DATA_PROCESSOR:
-                            request = Request.from_dict(data)
+                        request = Request.from_dict(data)
                         request.metrics.scheduler_recv_req_time = time.time()
                         main_process_metrics.requests_number.inc()
                         trace_carrier = data.get("trace_carrier")

fastdeploy/entrypoints/engine_client.py

@@ -437,7 +437,7 @@ async def add_requests(self, task):
     def _send_task(self, task):
         if envs.ZMQ_SEND_BATCH_DATA:
             task["zmq_worker_pid"] = self.worker_pid
-        if not self.enable_mm and not envs.ENABLE_V1_DATA_PROCESSOR:
+        if not self.enable_mm:
             self.zmq_client.send_json(task)
         else:
             if envs.FD_ENABLE_E2W_TENSOR_CONVERT:
@@ -599,7 +599,7 @@ async def run_control_method(self, request: ControlRequest):
         req_dict = request.to_dict()
         if envs.ZMQ_SEND_BATCH_DATA:
             req_dict["zmq_worker_pid"] = self.worker_pid
-        if not self.enable_mm and not envs.ENABLE_V1_DATA_PROCESSOR:
+        if not self.enable_mm:
             self.zmq_client.send_json(req_dict)
         else:
             self.zmq_client.send_pyobj(req_dict)

fastdeploy/entrypoints/openai/serving_chat.py

@@ -26,7 +26,7 @@
 
 import fastdeploy.envs as envs
 import fastdeploy.metrics.trace as tracing
-from fastdeploy.engine.request import Request, RequestOutput
+from fastdeploy.engine.request import RequestOutput
 from fastdeploy.entrypoints.openai.protocol import (
     ChatCompletionRequest,
     ChatCompletionResponse,
@@ -145,10 +145,7 @@ async def create_chat_completion(self, request: ChatCompletionRequest):
             prompt_tokens = None
             max_tokens = None
             try:
-                if not envs.ENABLE_V1_DATA_PROCESSOR:
-                    current_req_dict = request.to_dict_for_infer(f"{request_id}_0")
-                else:
-                    current_req_dict = Request.from_generic_request(request, request_id=f"{request_id}_0")
+                current_req_dict = request.to_dict_for_infer(f"{request_id}_0")
                 if "chat_template" not in current_req_dict:
                     current_req_dict["chat_template"] = self.chat_template
                 current_req_dict["metrics"]["arrival_time"] = time.time()

fastdeploy/entrypoints/openai/serving_completion.py

@@ -27,7 +27,7 @@
 
 import fastdeploy.envs as envs
 import fastdeploy.metrics.trace as tracing
-from fastdeploy.engine.request import Request, RequestOutput
+from fastdeploy.engine.request import RequestOutput
 from fastdeploy.entrypoints.openai.protocol import (
     CompletionLogprobs,
     CompletionRequest,
@@ -178,10 +178,7 @@ async def create_completion(self, request: CompletionRequest):
             try:
                 for idx, prompt in enumerate(request_prompts):
                     request_id_idx = f"{request_id}_{idx}"
-                    if not envs.ENABLE_V1_DATA_PROCESSOR:
-                        current_req_dict = request.to_dict_for_infer(request_id_idx, prompt)
-                    else:
-                        current_req_dict = Request.from_generic_request(request, request_id=f"{request_id}_0")
+                    current_req_dict = request.to_dict_for_infer(request_id_idx, prompt)
                     current_req_dict["metrics"]["arrival_time"] = time.time()
                     prompt_token_ids = await self.engine_client.format_and_add_data(current_req_dict)  # tokenize
                     if isinstance(prompt_token_ids, np.ndarray):

fastdeploy/entrypoints/openai/serving_embedding.py

@@ -15,20 +15,17 @@
 """
 
 import base64
-import time
 from collections.abc import AsyncGenerator
 from typing import Literal, Union
 
 import numpy as np
 from typing_extensions import assert_never, override
 
-import fastdeploy.envs as envs
 from fastdeploy.engine.pooling_params import PoolingParams
 from fastdeploy.engine.request import (
     EmbeddingOutput,
     EmbeddingRequestOutput,
     PoolingRequestOutput,
-    Request,
 )
 from fastdeploy.entrypoints.openai.protocol import (
     EmbeddingCompletionRequest,
@@ -69,25 +66,13 @@ def __init__(self, engine_client, models, cfg, pid, ips, max_waiting_time, chat_
     @override
     def _request_to_dict(self, ctx: ServeContext):
         request: EmbeddingRequest = ctx.request
-        if not envs.ENABLE_V1_DATA_PROCESSOR:
-            request_dict = super()._request_to_dict(ctx)
-            if hasattr(request, "to_pooling_params"):
-                pooling_params: PoolingParams = request.to_pooling_params()
-                pooling_params.verify("embed", self.cfg.model_config)
-                request_dict["pooling_params"] = pooling_params.to_dict()
-                request_dict["metrics"] = {}
-            return request_dict
-        else:
-            request_obj = None
-            if hasattr(request, "to_pooling_params"):
-                pooling_params: PoolingParams = request.to_pooling_params()
-                pooling_params.verify("embed", self.cfg.model_config)
-                request_obj = Request.from_generic_request(
-                    req=request, request_id=ctx.request_id, pooling_params=pooling_params
-                )
-                request_obj.metrics.arrival_time = time.time()
-                super()._process_chat_template_kwargs(request_obj)
-            return request_obj
+        request_dict = super()._request_to_dict(ctx)
+        if hasattr(request, "to_pooling_params"):
+            pooling_params: PoolingParams = request.to_pooling_params()
+            pooling_params.verify("embed", self.cfg.model_config)
+            request_dict["pooling_params"] = pooling_params.to_dict()
+            request_dict["metrics"] = {}
+        return request_dict
 
     @override
     def _request_to_batch_dicts(self, ctx: ServeContext):

fastdeploy/entrypoints/openai/serving_reward.py

@@ -14,14 +14,12 @@
 # limitations under the License.
 """
 
-import time
 from collections.abc import AsyncGenerator
 
 from typing_extensions import override
 
-import fastdeploy.envs as envs
 from fastdeploy.engine.pooling_params import PoolingParams
-from fastdeploy.engine.request import PoolingRequestOutput, Request, RewardRequestOutput
+from fastdeploy.engine.request import PoolingRequestOutput, RewardRequestOutput
 from fastdeploy.entrypoints.openai.protocol import (
     ChatRewardData,
     ChatRewardRequest,
@@ -46,25 +44,13 @@ def __init__(self, engine_client, models, cfg, pid, ips, max_waiting_time, chat_
     @override
     def _request_to_dict(self, ctx: ServeContext):
         request: ChatRewardRequest = ctx.request
-        if not envs.ENABLE_V1_DATA_PROCESSOR:
-            request_dict = super()._request_to_dict(ctx)
-            if hasattr(request, "to_pooling_params"):
-                pooling_params: PoolingParams = request.to_pooling_params()
-                pooling_params.verify("reward", self.cfg.model_config)
-                request_dict["pooling_params"] = pooling_params.to_dict()
-                request_dict["metrics"] = {}
-            return request_dict
-        else:
-            request_obj: Request = None
-            if hasattr(request, "to_pooling_params"):
-                pooling_params: PoolingParams = request.to_pooling_params()
-                pooling_params.verify("reward", self.cfg.model_config)
-                request_obj = Request.from_generic_request(
-                    req=request, request_id=ctx.request_id, pooling_params=pooling_params
-                )
-                request_obj.metrics.arrival_time = time.time()
-                super()._process_chat_template_kwargs(request_obj)
-            return request_obj
+        request_dict = super()._request_to_dict(ctx)
+        if hasattr(request, "to_pooling_params"):
+            pooling_params: PoolingParams = request.to_pooling_params()
+            pooling_params.verify("reward", self.cfg.model_config)
+            request_dict["pooling_params"] = pooling_params.to_dict()
+            request_dict["metrics"] = {}
+        return request_dict
 
     @override
     def _request_to_batch_dicts(self, ctx: ServeContext):

fastdeploy/envs.py

@@ -96,8 +96,6 @@ def _validate_split_kv_size(value: int) -> int:
     "EXPORTER_OTLP_HEADERS": lambda: os.getenv("EXPORTER_OTLP_HEADERS"),
     # enable kv cache block scheduler v1 (no need for kv_cache_ratio)
     "ENABLE_V1_KVCACHE_SCHEDULER": lambda: int(os.getenv("ENABLE_V1_KVCACHE_SCHEDULER", "1")),
-    # enable data processor v2
-    "ENABLE_V1_DATA_PROCESSOR": lambda: int(os.getenv("ENABLE_V1_DATA_PROCESSOR", "0")),
     # set prealloc block num for decoder
     "FD_ENC_DEC_BLOCK_NUM": lambda: int(os.getenv("FD_ENC_DEC_BLOCK_NUM", "2")),
     # enbale max prefill of one execute step

fastdeploy/input/preprocess.py

@@ -19,7 +19,6 @@
 from fastdeploy.config import ErnieArchitectures, ModelConfig
 from fastdeploy.entrypoints.openai.tool_parsers import ToolParserManager
 from fastdeploy.reasoning import ReasoningParserManager
-from fastdeploy.utils import envs
 from fastdeploy.utils import llm_logger as logger
 
 
@@ -82,39 +81,20 @@ def create_processor(self):
         except Exception as e:
             logger.info(f"Plugin input processor not available ({e}), using built-in processor")
             if not self.model_config.enable_mm:
-                if not envs.ENABLE_V1_DATA_PROCESSOR:
-                    from fastdeploy.input.text_processor import TextProcessor
-
-                    tokenizer_type = "ernie4_5" if ErnieArchitectures.contains_ernie_arch(architecture) else "auto"
-                    self.processor = TextProcessor(
-                        model_name_or_path=self.model_name_or_path,
-                        tokenizer_type=tokenizer_type,
-                        reasoning_parser_obj=reasoning_parser_obj,
-                        tool_parser_obj=tool_parser_obj,
-                    )
-                else:
-                    if not ErnieArchitectures.contains_ernie_arch(architecture):
-                        from fastdeploy.input.v1.text_processor import DataProcessor
-                    else:
-                        from fastdeploy.input.v1.ernie4_5_processor import (
-                            Ernie4_5Processor as DataProcessor,
-                        )
-
-                    self.processor = DataProcessor(
-                        model_name_or_path=self.model_name_or_path,
-                        reasoning_parser_obj=reasoning_parser_obj,
-                        tool_parser_obj=tool_parser_obj,
-                    )
+                from fastdeploy.input.text_processor import TextProcessor
+
+                tokenizer_type = "ernie4_5" if ErnieArchitectures.contains_ernie_arch(architecture) else "auto"
+                self.processor = TextProcessor(
+                    model_name_or_path=self.model_name_or_path,
+                    tokenizer_type=tokenizer_type,
+                    reasoning_parser_obj=reasoning_parser_obj,
+                    tool_parser_obj=tool_parser_obj,
+                )
             else:
                 if ErnieArchitectures.contains_ernie_arch(architecture):
-                    if not envs.ENABLE_V1_DATA_PROCESSOR:
-                        from fastdeploy.input.ernie4_5_vl_processor import (
-                            Ernie4_5_VLProcessor,
-                        )
-                    else:
-                        from fastdeploy.input.v1.ernie4_5_vl_processor import (
-                            Ernie4_5_VLProcessor,
-                        )
+                    from fastdeploy.input.ernie4_5_vl_processor import (
+                        Ernie4_5_VLProcessor,
+                    )
 
                     self.processor = Ernie4_5_VLProcessor(
                         model_name_or_path=self.model_name_or_path,
@@ -125,14 +105,9 @@ def create_processor(self):
                         enable_processor_cache=self.enable_processor_cache,
                     )
                 elif "PaddleOCRVL" in architecture:
-                    if not envs.ENABLE_V1_DATA_PROCESSOR:
-                        from fastdeploy.input.paddleocr_vl_processor import (
-                            PaddleOCRVLProcessor,
-                        )
-                    else:
-                        from fastdeploy.input.v1.paddleocr_vl_processor import (
-                            PaddleOCRVLProcessor,
-                        )
+                    from fastdeploy.input.paddleocr_vl_processor import (
+                        PaddleOCRVLProcessor,
+                    )
 
                     self.processor = PaddleOCRVLProcessor(
                         config=self.model_config,
@@ -142,12 +117,7 @@ def create_processor(self):
                         reasoning_parser_obj=reasoning_parser_obj,
                     )
                 elif "Qwen2_5_VL" in architecture:
-                    if not envs.ENABLE_V1_DATA_PROCESSOR:
-                        from fastdeploy.input.qwen_vl_processor import QwenVLProcessor
-                    else:
-                        from fastdeploy.input.v1.qwen_vl_processor import (
-                            QwenVLProcessor,
-                        )
+                    from fastdeploy.input.qwen_vl_processor import QwenVLProcessor
 
                     self.processor = QwenVLProcessor(
                         config=self.model_config,
@@ -158,12 +128,7 @@ def create_processor(self):
                         enable_processor_cache=self.enable_processor_cache,
                     )
                 elif "Qwen3VL" in architecture:
-                    if not envs.ENABLE_V1_DATA_PROCESSOR:
-                        from fastdeploy.input.qwen3_vl_processor import Qwen3VLProcessor
-                    else:
-                        from fastdeploy.input.v1.qwen3_vl_processor import (
-                            Qwen3VLProcessor,
-                        )
+                    from fastdeploy.input.qwen3_vl_processor import Qwen3VLProcessor
 
                     self.processor = Qwen3VLProcessor(
                         config=self.model_config,