feat: adding support for torch xpu device

[srini047]

Related Issues

• fixes #9039

Proposed Changes:

Add support for Intel xpu device

How did you test it?

Notes for the reviewer

I don't have a Intel XPU device. So unable to test it 😔

Checklist

• I have read the contributors guidelines and the code of conduct
• I have updated the related issue with new insights and changes
• I added unit tests and updated the docstrings
• I've used one of the conventional commit types for my PR title: fix:, feat:, build:, chore:, ci:, docs:, style:, refactor:, perf:, test: and added ! in case the PR includes breaking changes.
• I documented my code
• I ran pre-commit hooks and fixed any issue

[coveralls]

Pull Request Test Coverage Report for Build 15703017100

Details

• 0 of 0 changed or added relevant lines in 0 files are covered.
• 22 unchanged lines in 1 file lost coverage.
• Overall coverage decreased (-0.001%) to 90.143%

---

Files with Coverage Reduction	New Missed Lines	%
utils/device.py	22	89.72%
<!--	Total:	22

Totals
Change from base Build 15690472265:	-0.001%
Covered Lines:	11550
Relevant Lines:	12813

---

💛 - Coveralls

[sjrl]

Failing unit test fixed by this PR https://github.com/deepset-ai/haystack/pull/9475

[srini047]

Failing unit test fixed by this PR https://github.com/deepset-ai/haystack/pull/9475

Thanks @sjrl Can someone help with the review.

[davidsbatista]

@srini047 any chance you have to run this script on XPU hardware?

from typing import  Dict, Any

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.utils.device import ComponentDevice, Device

def test_device(device: ComponentDevice, device_name: str) -> Dict[str, Any]:
    result = {
        "device_name": device_name,
        "device_str": str(device.to_torch_str() if hasattr(device, '_single_device') and device._single_device else device),
        "success": False,
        "error": None,
        "embedding_shape": None
    }
    
    try:
        embedder = SentenceTransformersTextEmbedder(
            model="sentence-transformers/all-MiniLM-L6-v2",
            device=device,
            progress_bar=False
        )
        embedder.warm_up()
        test_text = "Hello world"
        embedding_result = embedder.run(text=test_text)
        result["success"] = True
        result["embedding_shape"] = len(embedding_result["embedding"]) if embedding_result["embedding"] else None
    except Exception as e:
        result["error"] = str(e)
        
    return result


def main():
    test_cases = [
        # (ComponentDevice.from_single(Device.cpu()), "CPU"),
        # (ComponentDevice.from_single(Device.gpu(0)), "GPU (CUDA:0)"),
        # (ComponentDevice.from_single(Device.gpu(1)), "GPU (CUDA:1)"),
        # (ComponentDevice.from_single(Device.mps()), "MPS (Apple Metal)"),
        (ComponentDevice.from_single(Device.xpu()), "XPU (Intel)")
    ]

    results = []
    for device, device_name in test_cases:
        print(f"\nTesting {device_name}...")
        result = test_device(device, device_name)
        results.append(result)
        if result["success"]:
            print(f"{device_name}: SUCCESS")
            print(result)
        else:
            print(f"{device_name}: FAILED")
            print(f"Error: {result['error']}")

    print("\n\n")
    successful = [r for r in results if r["success"]]
    failed = [r for r in results if not r["success"]]
    
    print(f"Successful devices")
    for result in successful:
        print(f"{result['device_name']}")
    if failed:
        print(f"\nFailed devices")
        for result in failed:
            print(f"{result['device_name']}: {result['error']}")

if __name__ == "__main__":
    main() 

[srini047]

@srini047 any chance you have to run this script on XPU hardware?

from typing import  Dict, Any

from haystack.components.embedders import SentenceTransformersTextEmbedder
from haystack.utils.device import ComponentDevice, Device

def test_device(device: ComponentDevice, device_name: str) -> Dict[str, Any]:
    result = {
        "device_name": device_name,
        "device_str": str(device.to_torch_str() if hasattr(device, '_single_device') and device._single_device else device),
        "success": False,
        "error": None,
        "embedding_shape": None
    }
    
    try:
        embedder = SentenceTransformersTextEmbedder(
            model="sentence-transformers/all-MiniLM-L6-v2",
            device=device,
            progress_bar=False
        )
        embedder.warm_up()
        test_text = "Hello world"
        embedding_result = embedder.run(text=test_text)
        result["success"] = True
        result["embedding_shape"] = len(embedding_result["embedding"]) if embedding_result["embedding"] else None
    except Exception as e:
        result["error"] = str(e)
        
    return result


def main():
    test_cases = [
        # (ComponentDevice.from_single(Device.cpu()), "CPU"),
        # (ComponentDevice.from_single(Device.gpu(0)), "GPU (CUDA:0)"),
        # (ComponentDevice.from_single(Device.gpu(1)), "GPU (CUDA:1)"),
        # (ComponentDevice.from_single(Device.mps()), "MPS (Apple Metal)"),
        (ComponentDevice.from_single(Device.xpu()), "XPU (Intel)")
    ]

    results = []
    for device, device_name in test_cases:
        print(f"\nTesting {device_name}...")
        result = test_device(device, device_name)
        results.append(result)
        if result["success"]:
            print(f"{device_name}: SUCCESS")
            print(result)
        else:
            print(f"{device_name}: FAILED")
            print(f"Error: {result['error']}")

    print("\n\n")
    successful = [r for r in results if r["success"]]
    failed = [r for r in results if not r["success"]]
    
    print(f"Successful devices")
    for result in successful:
        print(f"{result['device_name']}")
    if failed:
        print(f"\nFailed devices")
        for result in failed:
            print(f"{result['device_name']}: {result['error']}")

if __name__ == "__main__":
    main() 

@davidsbatista I don't have a XPU hardware. I have only MPS. So that's why unable to test it. Do you have any idea to access the XPU hardware for free for time being? I'm checking as well to test the code.