Retrospective: How We Scaled Our AI Inference Pipeline to 1M Requests/Second in 2026 with PyTorch 2.3

import torch
import torch.nn as nn
from torch.utils.data import DataLoader
import logging
from typing import Optional, Dict, Any
import os
from dataclasses import dataclass

# Configure logging for production tracing
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

@dataclass
class CompilationConfig:
    """Configuration for PyTorch 2.3 torch.compile workflow"""
    model_path: str
    batch_size: int = 32
    max_seq_len: int = 2048
    dtype: torch.dtype = torch.bfloat16
    compile_mode: str = "max-autotune"  # PyTorch 2.3 only: max-autotune, reduce-overhead, default
    dynamic_shape: bool = True
    cache_dir: str = "/tmp/torch_compile_cache"

class LLMCompiler:
    def __init__(self, config: CompilationConfig):
        self.config = config
        self.model: Optional[nn.Module] = None
        self.compiled_model: Optional[torch.ScriptModule] = None
        os.makedirs(self.config.cache_dir, exist_ok=True)
        torch._inductor.config.cache_dir = self.config.cache_dir

    def load_model(self) -> None:
        """Load pretrained 14B LLM from disk with error handling"""
        try:
            logger.info(f"Loading model from {self.config.model_path}")
            # Simulate loading a 14B parameter model - in production use HuggingFace Transformers or custom loader
            # For this example, we use a dummy LLM class matching our production architecture
            self.model = self._init_dummy_llm()
            self.model.to("cuda" if torch.cuda.is_available() else "cpu")
            self.model.eval()
            logger.info(f"Model loaded successfully. Parameters: {sum(p.numel() for p in self.model.parameters()):,}")
        except FileNotFoundError as e:
            logger.error(f"Model file not found: {e}")
            raise
        except RuntimeError as e:
            logger.error(f"CUDA OOM when loading model: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error loading model: {e}")
            raise

    def _init_dummy_llm(self) -> nn.Module:
        """Dummy 14B parameter LLM matching production architecture for example purposes"""
        # In production, this is a Llama 3 14B model from https://github.com/meta-llama/llama
        class DummyLLM(nn.Module):
            def __init__(self, hidden_size=5120, num_layers=40, num_heads=40, vocab_size=128256):
                super().__init__()
                self.embed = nn.Embedding(vocab_size, hidden_size)
                self.layers = nn.ModuleList([
                    nn.TransformerDecoderLayer(d_model=hidden_size, nhead=num_heads, dim_feedforward=hidden_size*4, batch_first=True)
                    for _ in range(num_layers)
                ])
                self.ln_f = nn.LayerNorm(hidden_size)
                self.head = nn.Linear(hidden_size, vocab_size)
            def forward(self, input_ids: torch.Tensor, attention_mask: Optional[torch.Tensor] = None):
                x = self.embed(input_ids)
                for layer in self.layers:
                    x = layer(x, src_mask=attention_mask)
                x = self.ln_f(x)
                return self.head(x)
        return DummyLLM()

    def compile_model(self) -> None:
        """Compile model with PyTorch 2.3 torch.compile, handle errors"""
        if self.model is None:
            raise ValueError("Model must be loaded before compilation")
        try:
            logger.info(f"Compiling model with mode: {self.config.compile_mode}, dynamic shape: {self.config.dynamic_shape}")
            # PyTorch 2.3 specific: enable dynamic shape specialization for variable sequence lengths
            compile_kwargs = {
                "mode": self.config.compile_mode,
                "dynamic": self.config.dynamic_shape,
                "fullgraph": True  # Enforce full graph capture for maximum throughput
            }
            self.compiled_model = torch.compile(self.model, **compile_kwargs)
            # Warmup run to trigger compilation and cache artifacts
            warmup_input = torch.randint(0, 128256, (self.config.batch_size, self.config.max_seq_len), device=self.model.device)
            _ = self.compiled_model(warmup_input)
            logger.info(f"Model compiled successfully. Cache stored at {self.config.cache_dir}")
        except torch._dynamo.exc.Unsupported as e:
            logger.error(f"Model contains unsupported operations for torch.compile: {e}")
            raise
        except Exception as e:
            logger.error(f"Compilation failed: {e}")
            raise

    def benchmark_throughput(self, num_iterations: int = 100) -> float:
        """Benchmark compiled model throughput in RPS"""
        if self.compiled_model is None:
            raise ValueError("Model must be compiled before benchmarking")
        self.model.eval()
        with torch.no_grad():
            total_requests = 0
            start = torch.cuda.Event(enable_timing=True)
            end = torch.cuda.Event(enable_timing=True)
            start.record()
            for _ in range(num_iterations):
                input_ids = torch.randint(0, 128256, (self.config.batch_size, self.config.max_seq_len), device=self.model.device)
                _ = self.compiled_model(input_ids)
                total_requests += self.config.batch_size
            end.record()
            torch.cuda.synchronize()
            elapsed = start.elapsed_time(end) / 1000  # Convert to seconds
            rps = total_requests / elapsed
            logger.info(f"Benchmark: {rps:.2f} RPS over {num_iterations} iterations")
            return rps

if __name__ == "__main__":
    # Production configuration for our 14B LLM workload
    config = CompilationConfig(
        model_path="/models/llama3-14b-chat",
        batch_size=32,
        max_seq_len=2048,
        compile_mode="max-autotune",
        dynamic_shape=True
    )
    compiler = LLMCompiler(config)
    try:
        compiler.load_model()
        compiler.compile_model()
        rps = compiler.benchmark_throughput()
        assert rps > 1000, f"Compiled model RPS {rps} below expected threshold"
    except Exception as e:
        logger.error(f"Pipeline failed: {e}")
        exit(1)

import torch
import torch.nn as nn
from torchserve import ModelHandler, Worker, KFService
from typing import List, Dict, Any, Optional
import logging
import time
import prometheus_client as prom
from dataclasses import dataclass
import os
import json

# Prometheus metrics for production monitoring
INFERENCE_RPS = prom.Counter("inference_requests_total", "Total inference requests")
INFERENCE_LATENCY = prom.Histogram("inference_latency_seconds", "Inference latency")
ERROR_RATE = prom.Counter("inference_errors_total", "Total inference errors")
GPU_UTIL = prom.Gauge("gpu_utilization_percent", "GPU utilization percentage")

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ServingConfig:
    """Configuration for PyTorch Serve 0.9.0 inference worker"""
    model_path: str
    batch_size: int = 32
    max_batch_delay_ms: int = 50  # Max time to wait for batch filling
    num_workers: int = 4
    dtype: torch.dtype = torch.bfloat16
    device: str = "cuda" if torch.cuda.is_available() else "cpu"
    metrics_port: int = 8080

class LLMHandler(ModelHandler):
    """Custom model handler for PyTorch Serve 0.9.0"""
    def __init__(self, config: ServingConfig):
        super().__init__()
        self.config = config
        self.model: Optional[nn.Module] = None
        self.tokenizer = None  # In production, use HuggingFace tokenizer
        self._load_model()

    def _load_model(self) -> None:
        """Load compiled model and tokenizer with error handling"""
        try:
            logger.info(f"Loading compiled model from {self.config.model_path}")
            # Load model saved after torch.compile in previous script
            self.model = torch.load(f"{self.config.model_path}/compiled_model.pt", map_location=self.config.device)
            self.model.eval()
            self.model.to(self.config.device)
            # Load tokenizer (dummy for example)
            self.tokenizer = lambda x: torch.randint(0, 128256, (len(x), 2048), device=self.config.device)
            logger.info("Model and tokenizer loaded successfully")
        except FileNotFoundError as e:
            logger.error(f"Model file missing: {e}")
            raise
        except RuntimeError as e:
            logger.error(f"Model load failed: {e}")
            raise

    def preprocess(self, inputs: List[Dict[str, Any]]) -> torch.Tensor:
        """Preprocess incoming requests into model inputs"""
        try:
            texts = [inp.get("text", "") for inp in inputs]
            if not texts:
                raise ValueError("No text provided in request")
            input_ids = self.tokenizer(texts)
            return input_ids
        except Exception as e:
            logger.error(f"Preprocessing failed: {e}")
            raise

    def inference(self, input_ids: torch.Tensor) -> torch.Tensor:
        """Run inference with compiled model, handle OOM and errors"""
        try:
            start = time.perf_counter()
            with torch.no_grad():
                outputs = self.model(input_ids)
            latency = time.perf_counter() - start
            INFERENCE_LATENCY.observe(latency)
            return outputs
        except torch.cuda.OutOfMemoryError as e:
            logger.error(f"CUDA OOM during inference: {e}")
            ERROR_RATE.inc()
            raise
        except Exception as e:
            logger.error(f"Inference failed: {e}")
            ERROR_RATE.inc()
            raise

    def postprocess(self, outputs: torch.Tensor, inputs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Convert model outputs to API responses"""
        try:
            # Dummy postprocessing: return top token ID
            top_tokens = outputs.argmax(dim=-1).cpu().tolist()
            return [{"output_token": top_tokens[i]} for i in range(len(inputs))]
        except Exception as e:
            logger.error(f"Postprocessing failed: {e}")
            raise

class LLMServeWorker(Worker):
    """Custom worker for batch inference with dynamic batching"""
    def __init__(self, config: ServingConfig):
        super().__init__()
        self.config = config
        self.handler = LLMHandler(config)
        self.batch_queue: List[Dict[str, Any]] = []
        self.last_batch_time = time.perf_counter()

    def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """Handle single request with batching logic"""
        INFERENCE_RPS.inc()
        self.batch_queue.append(request)
        # Wait for batch to fill or timeout
        if len(self.batch_queue) >= self.config.batch_size:
            return self._process_batch()
        elif (time.perf_counter() - self.last_batch_time) * 1000 >= self.config.max_batch_delay_ms:
            return self._process_batch()
        else:
            # For simplicity, process immediately in this example (production uses async batching)
            return self._process_batch()

    def _process_batch(self) -> Dict[str, Any]:
        """Process current batch of requests"""
        if not self.batch_queue:
            return {}
        batch = self.batch_queue.copy()
        self.batch_queue.clear()
        self.last_batch_time = time.perf_counter()
        try:
            input_ids = self.handler.preprocess(batch)
            outputs = self.handler.inference(input_ids)
            return self.handler.postprocess(outputs, batch)
        except Exception as e:
            logger.error(f"Batch processing failed: {e}")
            return {"error": str(e)}

if __name__ == "__main__":
    config = ServingConfig(
        model_path="/models/compiled-llama3-14b",
        batch_size=32,
        max_batch_delay_ms=50,
        num_workers=4
    )
    # Start Prometheus metrics server
    prom.start_http_server(config.metrics_port)
    # Initialize worker and start serving
    worker = LLMServeWorker(config)
    logger.info(f"Starting inference worker on device {config.device}")
    # In production, this integrates with PyTorch Serve 0.9.0's KFService
    # See https://github.com/pytorch/serve/releases/tag/v0.9.0 for full integration docs

import boto3
import torch
import os
import logging
from typing import Optional, List
import time
from dataclasses import dataclass
import json
from botocore.exceptions import ClientError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class SpotConfig:
    """Configuration for spot instance inference manager"""
    model_s3_bucket: str = "our-llm-models"
    model_s3_key: str = "llama3-14b/compiled"
    local_cache_dir: str = "/tmp/model_cache"
    health_check_interval: int = 10  # Seconds between preemption checks
    fallback_on_demand: bool = True
    region: str = "us-east-1"

class SpotInstanceManager:
    """Manage spot instances for inference with preemptible-aware caching"""
    def __init__(self, config: SpotConfig):
        self.config = config
        self.s3 = boto3.client("s3", region_name=config.region)
        self.ec2 = boto3.client("ec2", region_name=config.region)
        self.instance_id: Optional[str] = None
        self.model_loaded: bool = False
        os.makedirs(self.config.local_cache_dir, exist_ok=True)

    def get_instance_id(self) -> str:
        """Fetch current instance ID from EC2 metadata"""
        if self.instance_id:
            return self.instance_id
        try:
            # In production, use IMDSv2 to fetch instance ID
            import requests
            resp = requests.get("http://169.254.169.254/latest/meta-data/instance-id", timeout=2)
            self.instance_id = resp.text
            logger.info(f"Detected instance ID: {self.instance_id}")
            return self.instance_id
        except Exception as e:
            logger.error(f"Failed to fetch instance ID: {e}")
            raise

    def check_preemption(self) -> bool:
        """Check if spot instance is marked for preemption"""
        try:
            instance_id = self.get_instance_id()
            resp = self.ec2.describe_instances(InstanceIds=[instance_id])
            instance = resp["Reservations"][0]["Instances"][0]
            # Check for spot instance termination notice
            termination_notice = requests.get("http://169.254.169.254/latest/meta-data/spot/termination-time", timeout=1)
            if termination_notice.status_code == 200:
                logger.warning(f"Spot instance {instance_id} marked for preemption in {termination_notice.text}s")
                return True
            return False
        except requests.exceptions.Timeout:
            # No termination notice
            return False
        except Exception as e:
            logger.error(f"Preemption check failed: {e}")
            return False

    def cache_model_to_s3(self) -> None:
        """Cache compiled model to S3 for fast recovery"""
        try:
            local_model_path = os.path.join(self.config.local_cache_dir, "compiled_model.pt")
            if not os.path.exists(local_model_path):
                logger.warning("No local model to cache")
                return
            logger.info(f"Caching model to s3://{self.config.model_s3_bucket}/{self.config.model_s3_key}")
            self.s3.upload_file(local_model_path, self.config.model_s3_bucket, self.config.model_s3_key)
            logger.info("Model cached to S3 successfully")
        except ClientError as e:
            logger.error(f"S3 upload failed: {e}")
            raise
        except Exception as e:
            logger.error(f"Model caching failed: {e}")
            raise

    def load_model_from_cache(self) -> None:
        """Load model from S3 or local cache with fallback"""
        local_model_path = os.path.join(self.config.local_cache_dir, "compiled_model.pt")
        try:
            if os.path.exists(local_model_path):
                logger.info("Loading model from local cache")
                return
            logger.info(f"Downloading model from s3://{self.config.model_s3_bucket}/{self.config.model_s3_key}")
            self.s3.download_file(self.config.model_s3_bucket, self.config.model_s3_key, local_model_path)
            logger.info("Model downloaded from S3 successfully")
        except ClientError as e:
            if e.response["Error"]["Code"] == "404":
                logger.error("Model not found in S3, falling back to on-demand instance")
                if self.config.fallback_on_demand:
                    self._provision_on_demand()
            else:
                logger.error(f"S3 download failed: {e}")
                raise
        except Exception as e:
            logger.error(f"Model load failed: {e}")
            raise

    def _provision_on_demand(self) -> None:
        """Provision on-demand instance as fallback (simplified for example)"""
        logger.warning("Provisioning on-demand instance as fallback")
        # In production, use AWS SDK to launch on-demand instance and reroute traffic
        # This would integrate with our Kubernetes control plane at https://github.com/our-org/ai-control-plane
        pass

    def start_health_check(self) -> None:
        """Start background thread to check for preemption and cache model"""
        import threading
        def health_loop():
            while True:
                try:
                    if self.check_preemption():
                        logger.warning("Preemption detected, caching model and shutting down")
                        self.cache_model_to_s3()
                        # Deregister from load balancer, shutdown after cache
                        time.sleep(30)
                        exit(0)
                    time.sleep(self.config.health_check_interval)
                except Exception as e:
                    logger.error(f"Health check failed: {e}")
        thread = threading.Thread(target=health_loop, daemon=True)
        thread.start()
        logger.info("Started preemption health check loop")

if __name__ == "__main__":
    config = SpotConfig()
    manager = SpotInstanceManager(config)
    try:
        manager.get_instance_id()
        manager.load_model_from_cache()
        manager.start_health_check()
        # Keep main thread alive
        while True:
            time.sleep(60)
    except Exception as e:
        logger.error(f"Spot manager failed: {e}")
        exit(1)

# Minimal compilation snippet for production LLMs
import torch
from torch._inductor import config

config.cache_dir = "/persistent/compile_cache"
model = load_your_llm()  # Your model loading logic
compiled_model = torch.compile(
    model,
    mode="max-autotune",
    dynamic=True,
    fullgraph=True
)
# Warmup to trigger compilation
_ = compiled_model(torch.randint(0, 128256, (32, 2048), device="cuda"))

# PyTorch Serve 0.9.0 config for LLM workloads
# config.properties
model_store=/models
default_batch_size=32
max_batch_delay_ms=50
enable_dynamic_batching=true
enable_metrics=true
metrics_port=8080
device=cuda
# Load compiled model
load_model=compiled_llm

# Check for spot termination (production snippet)
import requests

def is_spot_terminating() -> bool:
    try:
        resp = requests.get(
            "http://169.254.169.254/latest/meta-data/spot/termination-time",
            timeout=1
        )
        return resp.status_code == 200
    except requests.exceptions.Timeout:
        return False