[LLM] support multi node deploy

Author: ltd0924

fastdeploy/engine/config.py

@@ -582,7 +582,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.is_master = True
         self._str_to_list("innode_prefill_ports", int)
         self._str_to_list("pod_ips", str)
 
@@ -641,6 +641,8 @@ def postprocess(self):
 
         if self.pod_ips is None:
             self.pod_ips = ["0.0.0.0"]
+        elif self.host_ip != self.pod_ips[0]:
+            self.is_master = False
 
         import paddle
         self.paddle_commit_id = paddle.version.commit
@@ -812,6 +814,9 @@ def reset_value(cls, value_name, key):
                     "return_full_hidden_states")
         reset_value(self.cache_config, "cache_dtype", "infer_model_dtype")
 
+    def _check_master(self):
+        return self.is_master
+    
     def _str_to_list(self, attr_name, default_type):
         if hasattr(self, attr_name):
             val = getattr(self, attr_name)

fastdeploy/entrypoints/llm.py

@@ -85,10 +85,16 @@ def __init__(
 
         self.mutex = threading.Lock()
         self.req_output = dict()
-
+        self.master_node_ip = self.llm_engine.config.pod_ips[0]
         self._receive_output_thread = threading.Thread(
             target=self._receive_output, daemon=True)
         self._receive_output_thread.start()
+    
+    def _check_master(self):
+        """
+        Check if the current node is the master node.
+        """
+        return self.llm_engine.config._check_master()
 
     def _receive_output(self):
         """
@@ -130,6 +136,10 @@ def generate(
             Union[str, list[str]]: The generated response.
         """
 
+        if not self._check_master():
+            err_msg = f"Only master node can accept completion request, please send request to master node: {self.master_node_ip}"
+            raise ValueError(err_msg)
+
         if sampling_params is None:
             sampling_params = self.default_sampling_params
 
@@ -182,6 +192,11 @@ def chat(
         Returns:
             Union[str, list[str]]: The generated response.
         """
+
+        if not self._check_master():
+            err_msg = f"Only master node can accept completion request, please send request to master node: {self.master_node_ip}"
+            raise ValueError(err_msg)
+        
         if sampling_params is None:
             sampling_params = self.default_sampling_params
 

fastdeploy/entrypoints/openai/api_server.py

@@ -120,8 +120,8 @@ async def lifespan(app: FastAPI):
                                  args.mm_processor_kwargs, args.enable_mm,
                                  args.reasoning_parser)
     app.state.dynamic_load_weight = args.dynamic_load_weight
-    chat_handler = OpenAIServingChat(engine_client, pid)
-    completion_handler = OpenAIServingCompletion(engine_client, pid)
+    chat_handler = OpenAIServingChat(engine_client, pid, args.pod_ips)
+    completion_handler = OpenAIServingCompletion(engine_client, pid, args.pod_ips)
     engine_client.create_zmq_client(model=pid, mode=zmq.PUSH)
     engine_client.pid = pid
     app.state.engine_client = engine_client

fastdeploy/entrypoints/openai/serving_chat.py

@@ -38,9 +38,7 @@
     ErrorResponse,
 )
 from fastdeploy.metrics.work_metrics import work_process_metrics
-
-from fastdeploy.utils import api_server_logger
-
+from fastdeploy.utils import api_server_logger, get_host_ip
 from fastdeploy.engine.request import RequestOutput
 
 
@@ -50,9 +48,18 @@ class OpenAIServingChat:
     OpenAI-style chat completions serving
     """
 
-    def __init__(self, engine_client, pid):
+    def __init__(self, engine_client, pid, pod_ips):
         self.engine_client = engine_client
         self.pid = pid
+        self.pod_ips = pod_ips
+        self.host_ip = get_host_ip()
+
+    def _check_master(self):
+        if self.pod_ips is None:
+            return True
+        if self.host_ip == self.pod_ips[0]:
+            return True
+        return False
 
     async def create_chat_completion(
         self,
@@ -61,6 +68,11 @@ async def create_chat_completion(
         """
         Create a new chat completion using the specified parameters.
         """
+
+        if not self._check_master():
+            err_msg = f"Only master node can accept completion request, please send request to master node: {self.pod_ips[0]}"
+            api_server_logger.error(err_msg)
+            return ErrorResponse(message=err_msg, code=400)
         if request.user is not None:
             request_id = f"chatcmpl-{request.user}-{uuid.uuid4()}"
         else:

fastdeploy/entrypoints/openai/serving_completion.py

@@ -39,19 +39,32 @@
     ToolCall,
     FunctionCall
 )
-from fastdeploy.utils import api_server_logger
+from fastdeploy.utils import api_server_logger, get_host_ip
 from fastdeploy.engine.request import RequestOutput
 
 
 class OpenAIServingCompletion:
-    def __init__(self, engine_client, pid):
+    def __init__(self, engine_client, pid, pod_ips):
         self.engine_client = engine_client
         self.pid = pid
+        self.pod_ips = pod_ips
+        self.host_ip = get_host_ip()
+
+    def _check_master(self):
+        if self.pod_ips is None:
+            return True
+        if self.host_ip == self.pod_ips[0]:
+            return True
+        return False
 
     async def create_completion(self, request: CompletionRequest):
         """
         Create a completion for the given prompt.
         """
+        if not self._check_master():
+            err_msg = f"Only master node can accept completion request, please send request to master node: {self.pod_ips[0]}"
+            api_server_logger.error(err_msg)
+            return ErrorResponse(message=err_msg, code=400)
         created_time = int(time.time())
         if request.user is not None:
             request_id = f"cmpl-{request.user}-{uuid.uuid4()}"