fix(qwen-asr): select Intel XPU devices

The Intel backend installs PyTorch XPU wheels, but Qwen ASR only
checked CUDA and MPS. Every Intel model therefore loaded on the CPU.

Select XPU when available and place the model on xpu:0. Keep the
existing CUDA, MPS, and CPU placement behavior.

Assisted-by: Codex:GPT-5 [apply_patch] [gh]

Author: localai-org-maint-bot

backend/python/qwen-asr/backend.py

@@ -18,6 +18,7 @@
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'common'))
 from grpc_auth import get_auth_interceptors
 from model_utils import resolve_model_reference
+from device_utils import device_map_for, select_device
 
 
 
@@ -95,13 +96,7 @@ def Health(self, request, context):
         return backend_pb2.Reply(message=bytes("OK", 'utf-8'))
 
     def LoadModel(self, request, context):
-        if torch.cuda.is_available():
-            device = "cuda"
-        else:
-            device = "cpu"
-        mps_available = hasattr(torch.backends, "mps") and torch.backends.mps.is_available()
-        if mps_available:
-            device = "mps"
+        device = select_device(torch)
         if not torch.cuda.is_available() and request.CUDA:
             return backend_pb2.Result(success=False, message="CUDA is not available")
 
@@ -123,7 +118,7 @@ def LoadModel(self, request, context):
         model_path, local_only = resolve_model_reference(
             request, "Qwen/Qwen3-ASR-1.7B"
         )
-        default_dtype = torch.bfloat16 if self.device == "cuda" else torch.float32
+        default_dtype = torch.bfloat16 if self.device in ("cuda", "xpu") else torch.float32
         load_dtype = default_dtype
         if "torch_dtype" in self.options:
             d = str(self.options["torch_dtype"]).lower()
@@ -145,12 +140,7 @@ def LoadModel(self, request, context):
         if attn_implementation is not None and isinstance(attn_implementation, str):
             attn_implementation = attn_implementation.strip() or None
 
-        if self.device == "mps":
-            device_map = None
-        elif self.device == "cuda":
-            device_map = "cuda:0"
-        else:
-            device_map = "cpu"
+        device_map = device_map_for(self.device)
 
         load_kwargs = dict(
             dtype=load_dtype,
@@ -423,4 +413,4 @@ def signal_handler(sig, frame):
     parser = argparse.ArgumentParser(description="Run the gRPC server.")
     parser.add_argument("--addr", default="localhost:50051", help="The address to bind the server to.")
     args = parser.parse_args()
-    serve(args.addr)
+    serve(args.addr)

backend/python/qwen-asr/device_utils.py

@@ -0,0 +1,18 @@
+def select_device(torch_module):
+    mps = getattr(getattr(torch_module, "backends", None), "mps", None)
+    if mps is not None and mps.is_available():
+        return "mps"
+    if torch_module.cuda.is_available():
+        return "cuda"
+    xpu = getattr(torch_module, "xpu", None)
+    if xpu is not None and xpu.is_available():
+        return "xpu"
+    return "cpu"
+
+
+def device_map_for(device):
+    if device == "mps":
+        return None
+    if device in ("cuda", "xpu"):
+        return f"{device}:0"
+    return "cpu"

backend/python/qwen-asr/device_utils_test.py

@@ -0,0 +1,58 @@
+import unittest
+
+from device_utils import device_map_for, select_device
+
+
+class Availability:
+    def __init__(self, available):
+        self._available = available
+
+    def is_available(self):
+        return self._available
+
+
+class TorchStub:
+    def __init__(self, *, cuda=False, mps=False, xpu=False):
+        self.cuda = Availability(cuda)
+        self.backends = type("Backends", (), {"mps": Availability(mps)})()
+        self.xpu = Availability(xpu)
+
+
+class SelectDeviceTest(unittest.TestCase):
+    def test_preserves_cuda_selection(self):
+        torch_module = TorchStub(cuda=True)
+
+        self.assertEqual(select_device(torch_module), "cuda")
+
+    def test_preserves_mps_selection(self):
+        torch_module = TorchStub(mps=True)
+
+        self.assertEqual(select_device(torch_module), "mps")
+
+    def test_selects_xpu_when_intel_gpu_is_available(self):
+        torch_module = TorchStub(xpu=True)
+
+        self.assertEqual(select_device(torch_module), "xpu")
+
+    def test_falls_back_to_cpu(self):
+        torch_module = TorchStub()
+
+        self.assertEqual(select_device(torch_module), "cpu")
+
+
+class DeviceMapTest(unittest.TestCase):
+    def test_preserves_cuda_model_placement(self):
+        self.assertEqual(device_map_for("cuda"), "cuda:0")
+
+    def test_preserves_mps_model_placement(self):
+        self.assertIsNone(device_map_for("mps"))
+
+    def test_places_the_model_on_the_first_xpu(self):
+        self.assertEqual(device_map_for("xpu"), "xpu:0")
+
+    def test_preserves_cpu_model_placement(self):
+        self.assertEqual(device_map_for("cpu"), "cpu")
+
+
+if __name__ == "__main__":
+    unittest.main()