FEAT: support replica removing (#4784)

Author: leslie2046

xinference/api/restful_api.py

@@ -787,6 +787,32 @@ async def terminate_model(self, model_uid: str) -> JSONResponse:
             raise HTTPException(status_code=500, detail=str(e))
         return JSONResponse(content=None)
 
+    async def terminate_model_replica(
+        self, model_uid: str, replica_id: int
+    ) -> JSONResponse:
+        try:
+            assert self._app is not None
+            remaining_replicas = await (
+                await self._get_supervisor_ref()
+            ).terminate_model_replica(model_uid, replica_id)
+            if remaining_replicas == 0:
+                self._app.router.routes = [
+                    route
+                    for route in self._app.router.routes
+                    if not (
+                        hasattr(route, "path")
+                        and isinstance(route.path, str)
+                        and route.path == "/" + model_uid
+                    )
+                ]
+            return JSONResponse(content={"remaining_replicas": remaining_replicas})
+        except ValueError as ve:
+            logger.error(str(ve), exc_info=True)
+            raise HTTPException(status_code=400, detail=str(ve))
+        except Exception as e:
+            logger.error(e, exc_info=True)
+            raise HTTPException(status_code=500, detail=str(e))
+
     async def _get_model_last_error(self, replica_model_uid: bytes, e: Exception):
         if not isinstance(e, xo.ServerClosed):
             return e

xinference/api/routers/models.py

@@ -113,6 +113,12 @@ def register_routes(api: "RESTfulAPI") -> None:
         methods=["GET"],
         dependencies=([Security(auth, scopes=["models:list"])] if is_auth else None),
     )
+    router.add_api_route(
+        "/v1/models/{model_uid}/replicas/{replica_id}",
+        api.terminate_model_replica,
+        methods=["DELETE"],
+        dependencies=([Security(auth, scopes=["models:stop"])] if is_auth else None),
+    )
     router.add_api_route(
         "/v1/models/{model_uid}/requests/{request_id}/abort",
         api.abort_request,

xinference/client/restful/async_restful_client.py

@@ -1432,6 +1432,21 @@ async def terminate_model(self, model_uid: str):
             )
         await _release_response(response)
 
+    async def terminate_model_replica(self, model_uid: str, replica_id: int) -> int:
+        """Terminate a specific replica of a running model."""
+
+        url = f"{self.base_url}/v1/models/{model_uid}/replicas/{replica_id}"
+
+        response = await self.session.delete(url, headers=self._headers)
+        if response.status != 200:
+            raise RuntimeError(
+                f"Failed to terminate model replica, detail: {await _get_error_string(response)}"
+            )
+
+        response_data = await response.json()
+        await _release_response(response)
+        return response_data["remaining_replicas"]
+
     async def get_launch_model_progress(self, model_uid: str) -> dict:
         """
         Get progress of the specific model.
@@ -1575,7 +1590,7 @@ async def get_model(self, model_uid: str) -> AsyncRESTfulModelHandle:
                 model_uid, self.base_url, auth_headers=self._headers
             )
         else:
-            raise ValueError(f"Unknown model type:{desc['model_type']}")
+            raise ValueError(f"Unknown model type: {desc['model_type']}")
 
     async def describe_model(self, model_uid: str):
         """

xinference/client/restful/restful_client.py

@@ -1329,6 +1329,20 @@ def terminate_model(self, model_uid: str):
                 f"Failed to terminate model, detail: {_get_error_string(response)}"
             )
 
+    def terminate_model_replica(self, model_uid: str, replica_id: int) -> int:
+        """Terminate a specific replica of a running model."""
+
+        url = f"{self.base_url}/v1/models/{model_uid}/replicas/{replica_id}"
+
+        response = self.session.delete(url, headers=self._headers)
+        if response.status_code != 200:
+            raise RuntimeError(
+                f"Failed to terminate model replica, detail: {_get_error_string(response)}"
+            )
+
+        response_data = response.json()
+        return response_data["remaining_replicas"]
+
     def get_launch_model_progress(self, model_uid: str) -> dict:
         """
         Get progress of the specific model.
@@ -1467,7 +1481,7 @@ def get_model(self, model_uid: str) -> RESTfulModelHandle:
                 model_uid, self.base_url, auth_headers=self._headers
             )
         else:
-            raise ValueError(f"Unknown model type:{desc['model_type']}")
+            raise ValueError(f"Unknown model type: {desc['model_type']}")
 
     def describe_model(self, model_uid: str):
         """

xinference/core/status_guard.py

@@ -143,3 +143,17 @@ def get_replica_statuses(self, model_uid: str) -> List[ReplicaStatus]:
         if model_uid in self._model_uid_to_info:
             return self._model_uid_to_info[model_uid].replica_statuses or []
         return []
+
+    def remove_replica_status(self, model_uid: str, replica_id: int) -> int:
+        """Remove status for a specific replica and return remaining replica count."""
+        if model_uid not in self._model_uid_to_info:
+            logger.warning(f"Model {model_uid} not found in status guard")
+            return 0
+
+        instance_info = self._model_uid_to_info[model_uid]
+        replica_statuses = instance_info.replica_statuses or []
+        instance_info.replica_statuses = [
+            status for status in replica_statuses if status.replica_id != replica_id
+        ]
+        instance_info.replica = len(instance_info.replica_statuses)
+        return instance_info.replica