Merge branch 'develop' into mm_structred_output

Author: Jiang-Jia-Jun

dockerfiles/Dockerfile.xpu

@@ -2,6 +2,9 @@ FROM ccr-2vdh3abv-pub.cnc.bj.baidubce.com/paddlepaddle/paddlenlp:llm-base-gcc12.
 
 WORKDIR /workspace
 
+ENV http_proxy=http://agent.baidu.com:8891
+ENV https_proxy=http://agent.baidu.com:8891
+
 RUN echo "\
 deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy main restricted universe multiverse \n\
 deb https://mirrors.tuna.tsinghua.edu.cn/ubuntu/ jammy-updates main restricted universe multiverse \n\
@@ -12,10 +15,10 @@ RUN apt-get update && apt-get install -y libibverbs-dev librdmacm-dev cmake pybi
 # uninstall existing package
 RUN python -m pip uninstall paddlepaddle-gpu paddlepaddle-xpu -y
 # install paddlepaddle
-RUN python -m pip install --no-cache-dir --progress-bar off --pre paddlepaddle-xpu -i https://www.paddlepaddle.org.cn/packages/nightly/xpu-p800/
+RUN python -m pip install --no-cache-dir --progress-bar off paddlepaddle-xpu==3.1.0 -i https://www.paddlepaddle.org.cn/packages/stable/xpu-p800/
 
 # get xtdk and xvllm and xre
-RUN mkdir -p /workspace/deps && cd /workspace/deps && wget https://klx-sdk-release-public.su.bcebos.com/xinfer/daily/eb/20250624/output.tar.gz && \
+RUN mkdir -p /workspace/deps && cd /workspace/deps && wget https://klx-sdk-release-public.su.bcebos.com/xinfer/daily/eb/20250710/output.tar.gz && \
     tar -zxf output.tar.gz  && mv output xvllm && \
     wget https://klx-sdk-release-public.su.bcebos.com/xtdk_15fusion/dev/3.2.40.1/xtdk-llvm15-ubuntu2004_x86_64.tar.gz && \
     tar -zxf xtdk-llvm15-ubuntu2004_x86_64.tar.gz && mv xtdk-llvm15-ubuntu2004_x86_64 xtdk && \
@@ -36,3 +39,4 @@ RUN cd /workspace/FastDeploy && bash build.sh && python -m pip install --no-cach
 
 ENV http_proxy=""
 ENV https_proxy=""
+ENV no_proxy=""

fastdeploy/download_model.py

@@ -142,12 +142,6 @@ def __init__(self, cfg):
                 disable_any_whitespace=self.cfg.disable_any_whitespace,
             )
 
-    def reset_scheduler(self):
-        """
-        Reset the scheduler to its initial state.
-        """
-        self.scheduler.reset()
-
     def start(self, api_server_pid=None):
         """
         Initializes the engine and starts its sub-services.

fastdeploy/engine/engine.py

@@ -32,7 +32,8 @@
                                                     ChatCompletionResponse,
                                                     CompletionRequest,
                                                     CompletionResponse,
-                                                    ErrorResponse)
+                                                    ErrorResponse,
+                                                    ControlSchedulerRequest)
 from fastdeploy.entrypoints.openai.serving_chat import OpenAIServingChat
 from fastdeploy.entrypoints.openai.serving_completion import \
     OpenAIServingCompletion
@@ -279,7 +280,7 @@ def launch_api_server() -> None:
     """
     if not is_port_available(args.host, args.port):
         raise Exception(f"The parameter `port`:{args.port} is already in use.")
-    
+
     api_server_logger.info(
         f"launch Fastdeploy api server... port: {args.port}")
     api_server_logger.info(f"args: {args.__dict__}")
@@ -326,7 +327,7 @@ def launch_metrics_server():
         raise Exception(
             f"The parameter `metrics_port`:{args.metrics_port} is already in use."
         )
-    
+
     prom_dir = cleanup_prometheus_files(True)
     os.environ["PROMETHEUS_MULTIPROC_DIR"] = prom_dir
     metrics_server_thread = threading.Thread(target=run_metrics_server,
@@ -347,10 +348,39 @@ def reset_scheduler():
 
     if llm_engine is None:
         return Response("Engine not loaded", status_code=500)
-    llm_engine.reset_scheduler()
+    llm_engine.scheduler.reset_scheduler()
     return Response("Scheduler Reset Successfully", status_code=200)
 
 
+@controller_app.post("/controller/scheduler")
+def control_scheduler(request: ControlSchedulerRequest):
+    """
+     Control the scheduler behavior with the given parameters.
+    """
+    content = ErrorResponse(object="", message="Scheduler updated successfully", code=0)
+    
+    global llm_engine
+    if llm_engine is None:
+        content.message = "Engine is not loaded"
+        content.code = 500
+        return JSONResponse(content=content.model_dump(), status_code=500)
+
+    if request.reset:
+        llm_engine.scheduler.reset_scheduler()
+
+    if request.load_shards_num or request.reallocate_shard:
+        if hasattr(llm_engine.scheduler, "update_config") and callable(llm_engine.scheduler.update_config):
+            llm_engine.scheduler.update_config(
+                load_shards_num=request.load_shards_num,
+                reallocate=request.reallocate_shard)
+        else:
+            content.message="This scheduler doesn't support the `update_config()` method."
+            content.code=400
+            return JSONResponse(content=content.model_dump(), status_code=400)
+
+    return JSONResponse(content=content.model_dump(), status_code=200)
+
+
 def run_controller_server():
     """
     run controller server
@@ -371,6 +401,11 @@ def launch_controller_server():
             f"The parameter `controller_port`:{args.controller_port} is already in use."
         )
 
+    if not is_port_available(args.host, args.controller_port):
+        raise Exception(
+            f"The parameter `controller_port`:{args.controller_port} is already in use."
+        )
+
     controller_server_thread = threading.Thread(target=run_controller_server,
                                                 daemon=True)
     controller_server_thread.start()

fastdeploy/entrypoints/openai/api_server.py

@@ -542,3 +542,12 @@ def check_logprobs(cls, data):
                 )
 
         return data
+      
+      
+class ControlSchedulerRequest(BaseModel):
+    """
+    Control scheduler request to the engine.
+    """
+    reset: Optional[bool] = False
+    load_shards_num: Optional[int] = None
+    reallocate_shard: Optional[bool] = False

fastdeploy/entrypoints/openai/protocol.py

@@ -58,6 +58,7 @@ def __init__(
         max_capture_batch_size: int = 64,
         guided_decoding_backend: str = "off",
         disable_any_whitespace: bool = True,
+        enable_logprob: bool = False,
     ):
         # Required parameters
         self.model_name_or_path = model_name_or_path
@@ -99,6 +100,7 @@ def __init__(
         self.max_capture_batch_size = max_capture_batch_size
         self.guided_decoding_backend = guided_decoding_backend
         self.disable_any_whitespace = disable_any_whitespace
+        self.enable_logprob = enable_logprob
 
     def __str__(self):
         return "\n".join(f"{k}: {v}" for k, v in self.__dict__.items())