Benchmark: PyTorch 2.4 vs. TensorFlow 2.17: Image Classification Accuracy 2026

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms, models
from torch.utils.data import DataLoader
import argparse
import logging
import os
import sys
from typing import Tuple

# Configure logging for training telemetry
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='PyTorch 2.4 ResNet-152 ImageNet-2026 Training')
    parser.add_argument('--data-dir', type=str, default='/data/imagenet-2026', help='Path to ImageNet-2026 dataset')
    parser.add_argument('--epochs', type=int, default=100, help='Number of training epochs')
    parser.add_argument('--batch-size', type=int, default=256, help='Batch size per GPU')
    parser.add_argument('--lr', type=float, default=0.1, help='Initial learning rate')
    parser.add_argument('--num-workers', type=int, default=16, help='DataLoader worker processes')
    parser.add_argument('--checkpoint-dir', type=str, default='./checkpoints', help='Checkpoint save directory')
    return parser.parse_args()

def get_data_loaders(args: argparse.Namespace) -> Tuple[DataLoader, DataLoader]:
    \"\"\"Initialize ImageNet-2026 train and validation DataLoaders with standard transforms\"\"\"
    train_transform = transforms.Compose([
        transforms.RandomResizedCrop(224),
        transforms.RandomHorizontalFlip(),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    ])
    val_transform = transforms.Compose([
        transforms.Resize(256),
        transforms.CenterCrop(224),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
    ])

    try:
        train_dataset = datasets.ImageFolder(os.path.join(args.data_dir, 'train'), transform=train_transform)
        val_dataset = datasets.ImageFolder(os.path.join(args.data_dir, 'val'), transform=val_transform)
        logger.info(f'Loaded {len(train_dataset)} training samples, {len(val_dataset)} validation samples')
    except Exception as e:
        logger.error(f'Failed to load dataset: {e}')
        sys.exit(1)

    train_loader = DataLoader(
        train_dataset,
        batch_size=args.batch_size,
        shuffle=True,
        num_workers=args.num_workers,
        pin_memory=True,
        persistent_workers=True
    )
    val_loader = DataLoader(
        val_dataset,
        batch_size=args.batch_size,
        shuffle=False,
        num_workers=args.num_workers,
        pin_memory=True
    )
    return train_loader, val_loader

def main():
    args = parse_args()
    # Verify CUDA availability
    if not torch.cuda.is_available():
        logger.error('CUDA is not available. Exiting.')
        sys.exit(1)
    device = torch.device('cuda')
    logger.info(f'Using device: {device}, GPU count: {torch.cuda.device_count()}')

    # Create checkpoint directory
    os.makedirs(args.checkpoint_dir, exist_ok=True)

    # Initialize model: ResNet-152 with 21k output classes for ImageNet-2026
    try:
        model = models.resnet152(weights=models.ResNet152_Weights.IMAGENET1K_V2)
        # Modify final fully connected layer for 21k classes
        in_features = model.fc.in_features
        model.fc = nn.Linear(in_features, 21000)
        model = model.to(device)
        # Wrap model in DistributedDataParallel if multiple GPUs
        if torch.cuda.device_count() > 1:
            model = nn.parallel.DistributedDataParallel(model)
        logger.info(f'Initialized ResNet-152 model with 21000 output classes')
    except Exception as e:
        logger.error(f'Failed to initialize model: {e}')
        sys.exit(1)

    # Initialize loss, optimizer, scheduler
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.SGD(model.parameters(), lr=args.lr, momentum=0.9, weight_decay=1e-4)
    scheduler = optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=args.epochs)

    # Training loop
    best_accuracy = 0.0
    for epoch in range(args.epochs):
        model.train()
        train_loss = 0.0
        for batch_idx, (inputs, labels) in enumerate(train_loader):
            inputs, labels = inputs.to(device), labels.to(device)
            optimizer.zero_grad()
            try:
                outputs = model(inputs)
                loss = criterion(outputs, labels)
                loss.backward()
                optimizer.step()
            except RuntimeError as e:
                if 'out of memory' in str(e):
                    logger.warning(f'OOM on batch {batch_idx}, skipping batch')
                    torch.cuda.empty_cache()
                    continue
                else:
                    raise e
            train_loss += loss.item()
            if batch_idx % 100 == 0:
                logger.info(f'Epoch {epoch+1}/{args.epochs}, Batch {batch_idx}/{len(train_loader)}, Loss: {loss.item():.4f}')

        # Validation loop
        model.eval()
        correct = 0
        total = 0
        val_loss = 0.0
        with torch.no_grad():
            for inputs, labels in val_loader:
                inputs, labels = inputs.to(device), labels.to(device)
                outputs = model(inputs)
                loss = criterion(outputs, labels)
                val_loss += loss.item()
                _, predicted = torch.max(outputs.data, 1)
                total += labels.size(0)
                correct += (predicted == labels).sum().item()
        accuracy = 100 * correct / total
        logger.info(f'Epoch {epoch+1}: Validation Accuracy: {accuracy:.2f}%, Train Loss: {train_loss/len(train_loader):.4f}, Val Loss: {val_loss/len(val_loader):.4f}')

        # Save best model
        if accuracy > best_accuracy:
            best_accuracy = accuracy
            torch.save(model.state_dict(), os.path.join(args.checkpoint_dir, 'best_model.pth'))
            logger.info(f'Saved best model with accuracy {best_accuracy:.2f}%')

        scheduler.step()

    logger.info(f'Training complete. Best validation accuracy: {best_accuracy:.2f}%')

if __name__ == '__main__':
    main()

import tensorflow as tf
from tensorflow.keras import layers, models, optimizers, callbacks
import argparse
import logging
import os
import sys
from typing import Tuple

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='TensorFlow 2.17 ResNet-152 ImageNet-2026 Training')
    parser.add_argument('--data-dir', type=str, default='/data/imagenet-2026', help='Path to ImageNet-2026 dataset')
    parser.add_argument('--epochs', type=int, default=100, help='Number of training epochs')
    parser.add_argument('--batch-size', type=int, default=256, help='Batch size per GPU')
    parser.add_argument('--lr', type=float, default=0.1, help='Initial learning rate')
    parser.add_argument('--num-workers', type=int, default=16, help='DataLoader worker processes')
    parser.add_argument('--checkpoint-dir', type=str, default='./checkpoints', help='Checkpoint save directory')
    return parser.parse_args()

def get_data_loaders(args: argparse.Namespace) -> Tuple[tf.data.Dataset, tf.data.Dataset]:
    \"\"\"Initialize ImageNet-2026 train and validation tf.data pipelines\"\"\"
    def preprocess_train(image_path: str, label: int) -> Tuple[tf.Tensor, int]:
        image = tf.io.read_file(image_path)
        image = tf.image.decode_jpeg(image, channels=3)
        image = tf.image.random_resized_crop(image, size=(224, 224))
        image = tf.image.random_flip_left_right(image)
        image = tf.cast(image, tf.float32) / 255.0
        image = tf.image.normalize(image, mean=[0.485, 0.456, 0.406], variance=[0.229**2, 0.224**2, 0.225**2])
        return image, label

    def preprocess_val(image_path: str, label: int) -> Tuple[tf.Tensor, int]:
        image = tf.io.read_file(image_path)
        image = tf.image.decode_jpeg(image, channels=3)
        image = tf.image.resize(image, (256, 256))
        image = tf.image.central_crop(image, central_fraction=224/256)
        image = tf.cast(image, tf.float32) / 255.0
        image = tf.image.normalize(image, mean=[0.485, 0.456, 0.406], variance=[0.229**2, 0.224**2, 0.225**2])
        return image, label

    try:
        # Load dataset using tf.data from directory structure matching ImageFolder
        train_ds = tf.keras.utils.image_dataset_from_directory(
            os.path.join(args.data_dir, 'train'),
            labels='inferred',
            label_mode='int',
            batch_size=args.batch_size,
            image_size=(224, 224),
            shuffle=True,
            seed=42
        )
        val_ds = tf.keras.utils.image_dataset_from_directory(
            os.path.join(args.data_dir, 'val'),
            labels='inferred',
            label_mode='int',
            batch_size=args.batch_size,
            image_size=(224, 224),
            shuffle=False,
            seed=42
        )
        # Apply preprocessing and performance optimizations
        train_ds = train_ds.map(preprocess_train, num_parallel_calls=tf.data.AUTOTUNE).prefetch(tf.data.AUTOTUNE)
        val_ds = val_ds.map(preprocess_val, num_parallel_calls=tf.data.AUTOTUNE).prefetch(tf.data.AUTOTUNE)
        logger.info(f'Loaded training dataset with {train_ds.cardinality()}, validation with {val_ds.cardinality()}')
    except Exception as e:
        logger.error(f'Failed to load dataset: {e}')
        sys.exit(1)
    return train_ds, val_ds

def main():
    args = parse_args()
    # Verify GPU availability
    gpus = tf.config.list_physical_devices('GPU')
    if not gpus:
        logger.error('No GPUs available. Exiting.')
        sys.exit(1)
    logger.info(f'Using {len(gpus)} GPUs: {gpus}')

    # Create checkpoint directory
    os.makedirs(args.checkpoint_dir, exist_ok=True)

    # Initialize model: ResNet-152 with 21k classes
    try:
        # Load pretrained ResNet152 from TensorFlow Hub
        inputs = layers.Input(shape=(224, 224, 3))
        # Use tf.keras.applications.ResNet152 with ImageNet1K V2 weights
        base_model = tf.keras.applications.ResNet152(weights='imagenet', include_top=False, input_tensor=inputs)
        # Add custom head for 21k classes
        x = layers.GlobalAveragePooling2D()(base_model.output)
        x = layers.Dense(21000, activation='softmax')(x)
        model = models.Model(inputs=inputs, outputs=x)
        logger.info(f'Initialized ResNet-152 model with 21000 output classes')
    except Exception as e:
        logger.error(f'Failed to initialize model: {e}')
        sys.exit(1)

    # Compile model with SGD, cosine annealing scheduler
    lr_schedule = optimizers.schedules.CosineDecay(
        initial_learning_rate=args.lr,
        decay_steps=args.epochs * 1000  # Approximate steps per epoch
    )
    optimizer = optimizers.SGD(learning_rate=lr_schedule, momentum=0.9, weight_decay=1e-4)
    model.compile(
        optimizer=optimizer,
        loss='sparse_categorical_crossentropy',
        metrics=['accuracy']
    )

    # Define callbacks
    checkpoint_cb = callbacks.ModelCheckpoint(
        filepath=os.path.join(args.checkpoint_dir, 'best_model.keras'),
        monitor='val_accuracy',
        mode='max',
        save_best_only=True
    )
    early_stopping_cb = callbacks.EarlyStopping(
        monitor='val_accuracy',
        patience=10,
        mode='max'
    )

    # Train model
    try:
        history = model.fit(
            train_ds,
            epochs=args.epochs,
            validation_data=val_ds,
            callbacks=[checkpoint_cb, early_stopping_cb],
            verbose=1
        )
        logger.info(f'Training complete. Best validation accuracy: {max(history.history[\"val_accuracy\"]):.2f}%')
    except RuntimeError as e:
        if 'out of memory' in str(e):
            logger.warning(f'OOM error during training: {e}')
            tf.config.experimental.reset_memory_stats(gpus[0])
        else:
            raise e
    except Exception as e:
        logger.error(f'Training failed: {e}')
        sys.exit(1)

if __name__ == '__main__':
    main()

import torch
import tensorflow as tf
import numpy as np
import time
import argparse
import logging
from typing import List, Dict

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description='PyTorch vs TensorFlow Inference Benchmark')
    parser.add_argument('--pytorch-model', type=str, default='./checkpoints/best_model.pth', help='PyTorch model path')
    parser.add_argument('--tf-model', type=str, default='./checkpoints/best_model.keras', help='TensorFlow model path')
    parser.add_argument('--batch-sizes', type=int, nargs='+', default=[1, 16, 32, 64], help='Batch sizes to test')
    parser.add_argument('--num-warmup', type=int, default=10, help='Number of warmup inference runs')
    parser.add_argument('--num-runs', type=int, default=100, help='Number of timed inference runs')
    return parser.parse_args()

def benchmark_pytorch(model_path: str, batch_size: int, num_warmup: int, num_runs: int) -> Dict:
    \"\"\"Benchmark PyTorch model inference\"\"\"
    logger.info(f'Benchmarking PyTorch model: {model_path}, batch size: {batch_size}')
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    try:
        # Load model
        model = torch.hub.load('pytorch/vision', 'resnet152', weights=None)
        in_features = model.fc.in_features
        model.fc = torch.nn.Linear(in_features, 21000)
        model.load_state_dict(torch.load(model_path, map_location=device))
        model = model.to(device)
        model.eval()
        logger.info(f'Loaded PyTorch model to {device}')
    except Exception as e:
        logger.error(f'Failed to load PyTorch model: {e}')
        return {}

    # Generate dummy input
    dummy_input = torch.randn(batch_size, 3, 224, 224).to(device)
    # Warmup runs
    with torch.no_grad():
        for _ in range(num_warmup):
            _ = model(dummy_input)
    torch.cuda.synchronize()

    # Timed runs
    latencies = []
    with torch.no_grad():
        for _ in range(num_runs):
            start = time.perf_counter()
            _ = model(dummy_input)
            torch.cuda.synchronize()
            end = time.perf_counter()
            latencies.append((end - start) * 1000)  # ms

    # Calculate metrics
    avg_latency = np.mean(latencies)
    p99_latency = np.percentile(latencies, 99)
    throughput = (batch_size * num_runs) / (sum(latencies) / 1000)  # images/sec
    return {
        'framework': 'PyTorch 2.4',
        'batch_size': batch_size,
        'avg_latency_ms': round(avg_latency, 2),
        'p99_latency_ms': round(p99_latency, 2),
        'throughput_img_per_sec': round(throughput, 2)
    }

def benchmark_tensorflow(model_path: str, batch_size: int, num_warmup: int, num_runs: int) -> Dict:
    \"\"\"Benchmark TensorFlow model inference\"\"\"
    logger.info(f'Benchmarking TensorFlow model: {model_path}, batch size: {batch_size}')
    try:
        # Load model
        model = tf.keras.models.load_model(model_path)
        logger.info(f'Loaded TensorFlow model')
    except Exception as e:
        logger.error(f'Failed to load TensorFlow model: {e}')
        return {}

    # Generate dummy input
    dummy_input = tf.random.normal((batch_size, 224, 224, 3))
    # Warmup runs
    for _ in range(num_warmup):
        _ = model(dummy_input)

    # Timed runs
    latencies = []
    for _ in range(num_runs):
        start = time.perf_counter()
        _ = model(dummy_input)
        end = time.perf_counter()
        latencies.append((end - start) * 1000)  # ms

    # Calculate metrics
    avg_latency = np.mean(latencies)
    p99_latency = np.percentile(latencies, 99)
    throughput = (batch_size * num_runs) / (sum(latencies) / 1000)  # images/sec
    return {
        'framework': 'TensorFlow 2.17',
        'batch_size': batch_size,
        'avg_latency_ms': round(avg_latency, 2),
        'p99_latency_ms': round(p99_latency, 2),
        'throughput_img_per_sec': round(throughput, 2)
    }

def main():
    args = parse_args()
    results: List[Dict] = []
    for batch_size in args.batch_sizes:
        # Benchmark PyTorch
        pt_result = benchmark_pytorch(args.pytorch_model, batch_size, args.num_warmup, args.num_runs)
        if pt_result:
            results.append(pt_result)
        # Benchmark TensorFlow
        tf_result = benchmark_tensorflow(args.tf_model, batch_size, args.num_warmup, args.num_runs)
        if tf_result:
            results.append(tf_result)
    # Print results table
    logger.info('\nInference Benchmark Results:')
    logger.info(f'{\"Framework\":<15} {\"Batch Size\":<10} {\"Avg Latency (ms)\":<15} {\"P99 Latency (ms)\":<15} {\"Throughput (img/s)\":<20}')
    for res in results:
        logger.info(f'{res[\"framework\"]:<15} {res[\"batch_size\"]:<10} {res[\"avg_latency_ms\"]:<15} {res[\"p99_latency_ms\"]:<15} {res[\"throughput_img_per_sec\"]:<20}')

if __name__ == '__main__':
    main()