Python for AI Engineering: PyTorch, CUDA & Async Orchestration

import asyncio
import torch
from typing import AsyncGenerator, List

class AsyncTensorInferenceEngine:
  def __init__(self, model_path: str):
      self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
      # Load compiled PyTorch model using TorchScript / Inductor
      self.model = torch.jit.load(model_path).to(self.device).eval()
      self.cuda_stream = torch.cuda.Stream(device=self.device)

  async def infer_batch_async(self, input_ids: List[List[int]]) -> torch.Tensor:
      loop = asyncio.get_running_loop()
      
      # Allocate pinned CPU memory for fast DMA transfer to GPU
      cpu_tensor = torch.tensor(input_ids, dtype=torch.long).pin_memory()
      
      def _gpu_compute():
          with torch.cuda.stream(self.cuda_stream):
              gpu_tensor = cpu_tensor.to(self.device, non_blocking=True)
              with torch.no_grad():
                  logits = self.model(gpu_tensor)
              self.cuda_stream.synchronize()
              return logits.cpu()

      # Offload GPU launch to threadpool worker to keep asyncio loop free
      return await loop.run_in_executor(None, _gpu_compute)

async def main():
  engine = AsyncTensorInferenceEngine("./models/compiled_transformer.pt")
  dummy_input = [[101, 2054, 2003, 1037, 13988, 102]]
  logits = await engine.infer_batch_async(dummy_input)
  print(f"Inference Logits Shape: {logits.shape}")

if __name__ == "__main__":
  asyncio.run(main())