The Performance Battle migration in TensorRT vs Llama 4: The Truth

import tensorrt as trt
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
import numpy as np
import os
import sys
from pathlib import Path

# TensorRT logger callback
class TRTLogger(trt.ILogger):
    def __init__(self, level=trt.Logger.VERBOSE):
        super().__init__(level)
        self.logs = []

    def log(self, severity, msg):
        self.logs.append(f"[{severity}] {msg}")
        if severity <= trt.Logger.WARNING:
            print(f"TRT {severity}: {msg}")

def build_llama4_tensorrt_engine(
    model_path: str = "meta-llama/Llama-4-8B-Instruct",
    engine_path: str = "./llama4_8b_fp8.trt",
    precision: str = "fp8",
    batch_size: int = 1,
    max_seq_len: int = 2048,
    workspace_size: int = 8 * (1 << 30)  # 8GB workspace
):
    """Builds a TensorRT engine for Llama 4 8B with specified precision.
    Supports fp16, fp8, int4 quantization."""
    logger = TRTLogger()
    # Check if model path exists
    if not Path(model_path).exists() and "meta-llama" in model_path:
        print(f"Downloading {model_path} from Hugging Face... (requires auth)")
        try:
            tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
            model = AutoModelForCausalLM.from_pretrained(
                model_path,
                torch_dtype=torch.float16,
                device_map="auto",
                trust_remote_code=True
            )
        except Exception as e:
            print(f"Failed to load model: {e}")
            sys.exit(1)
    else:
        tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
        model = AutoModelForCausalLM.from_pretrained(model_path, trust_remote_code=True)

    # Validate precision
    supported_precisions = ["fp16", "fp8", "int4"]
    if precision not in supported_precisions:
        raise ValueError(f"Unsupported precision {precision}. Use {supported_precisions}")

    # Initialize TensorRT builder
    builder = trt.Builder(logger)
    network = builder.create_network(1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH))
    config = builder.create_builder_config()
    config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, workspace_size)

    # Enable quantization if needed
    if precision == "fp8":
        if not builder.platform_has_fast_fp8():
            print("Warning: Platform does not support fast FP8, falling back to FP16")
            precision = "fp16"
        else:
            config.set_flag(trt.BuilderFlag.FP8)
            config.set_flag(trt.BuilderFlag.OBEY_PRECISION_CONSTRAINTS)
    elif precision == "int4":
        config.set_flag(trt.BuilderFlag.INT4)
        # Enable AWQ quantization (requires TensorRT 10.2+)
        config.quantization_config = trt.QuantizationConfig()
        config.quantization_config.set_calibration_profile(None)  # Use default calibration

    # Define input shapes
    input_ids_shape = (batch_size, max_seq_len)
    attention_mask_shape = (batch_size, max_seq_len)

    # Add inputs to network
    input_ids = network.add_input(name="input_ids", dtype=trt.int32, shape=input_ids_shape)
    attention_mask = network.add_input(name="attention_mask", dtype=trt.int32, shape=attention_mask_shape)

    # Export model to ONNX first (simplified for example; production uses direct conversion)
    onnx_path = "./llama4_8b.onnx"
    if not Path(onnx_path).exists():
        print("Exporting model to ONNX...")
        try:
            torch.onnx.export(
                model,
                (torch.randint(0, 100, input_ids_shape), torch.ones(attention_mask_shape, dtype=torch.int32)),
                onnx_path,
                input_names=["input_ids", "attention_mask"],
                output_names=["logits"],
                dynamic_axes={
                    "input_ids": {0: "batch_size", 1: "seq_len"},
                    "attention_mask": {0: "batch_size", 1: "seq_len"},
                    "logits": {0: "batch_size", 1: "seq_len"}
                },
                opset_version=17
            )
        except Exception as e:
            print(f"ONNX export failed: {e}")
            sys.exit(1)

    # Parse ONNX model
    parser = trt.OnnxParser(network, logger)
    with open(onnx_path, "rb") as f:
        onnx_data = f.read()
    if not parser.parse(onnx_data):
        print("Failed to parse ONNX model:")
        for i in range(parser.num_errors):
            print(parser.get_error(i))
        sys.exit(1)

    # Build engine
    print(f"Building TensorRT engine for {model_path} ({precision})...")
    try:
        engine = builder.build_engine(network, config)
    except Exception as e:
        print(f"Engine build failed: {e}")
        sys.exit(1)

    if engine is None:
        print("Engine build returned None")
        sys.exit(1)

    # Serialize engine to disk
    with open(engine_path, "wb") as f:
        f.write(engine.serialize())
    print(f"Engine saved to {engine_path}")
    return engine

if __name__ == "__main__":
    # Example usage: Build FP8 engine for batch size 1, max seq len 2048
    try:
        build_llama4_tensorrt_engine(
            model_path="meta-llama/Llama-4-8B-Instruct",
            engine_path="./llama4_8b_fp8_bs1.trt",
            precision="fp8",
            batch_size=1,
            max_seq_len=2048
        )
    except Exception as e:
        print(f"Fatal error: {e}")
        sys.exit(1)

import os
import sys
import time
import torch
from pathlib import Path
from llama.models.llama4 import Llama4, Llama4Config
from llama.tokenizer import Tokenizer
from llama.generator import Generator, GeneratorArgs
import numpy as np

# Set environment variables for reproducibility
os.environ["CUDA_VISIBLE_DEVICES"] = "0"
os.environ["PYTHONHASHSEED"] = "42"
torch.manual_seed(42)

def run_llama4_reference_inference(
    model_dir: str = "./llama4_models/8B",
    tokenizer_path: str = "./llama4_models/tokenizer.model",
    prompt: str = "Explain quantum computing in 3 sentences.",
    max_seq_len: int = 2048,
    max_gen_len: int = 128,
    temperature: float = 0.7,
    top_p: float = 0.9,
    batch_size: int = 1
):
    """Runs inference using the official Llama 4 reference implementation.
    Supports FP16, FP8 (via torchao), AWQ INT4 quantization."""
    # Validate model directory exists
    if not Path(model_dir).exists():
        print(f"Error: Model directory {model_dir} not found. Download from https://github.com/meta/llama")
        sys.exit(1)
    if not Path(tokenizer_path).exists():
        print(f"Error: Tokenizer file {tokenizer_path} not found.")
        sys.exit(1)

    # Load tokenizer
    try:
        tokenizer = Tokenizer(tokenizer_path)
    except Exception as e:
        print(f"Failed to load tokenizer: {e}")
        sys.exit(1)

    # Load model config
    config_path = Path(model_dir) / "params.json"
    if not config_path.exists():
        print(f"Error: Config file {config_path} not found.")
        sys.exit(1)
    config = Llama4Config.from_json(config_path)
    config.max_seq_len = max_seq_len

    # Load model with FP8 quantization (using torchao)
    print(f"Loading Llama 4 model from {model_dir}...")
    try:
        model = Llama4(config)
        if torch.cuda.is_available():
            model = model.cuda()
        # Apply FP8 quantization if CUDA supports it
        if torch.cuda.is_bf16_supported() and hasattr(torch, "ao"):
            import torchao
            torchao.quantization.quantize_(
                model,
                torchao.quantization.Fp8WeightOnlyConfig()
            )
            print("Applied FP8 weight-only quantization via torchao")
    except Exception as e:
        print(f"Failed to load model: {e}")
        sys.exit(1)

    # Initialize generator
    generator = Generator(model, tokenizer)

    # Prepare batch of prompts
    prompts = [prompt] * batch_size
    tokens = [tokenizer.encode(p, bos=True, eos=False) for p in prompts]

    # Pad tokens to max_seq_len
    padded_tokens = []
    for t in tokens:
        if len(t) > max_seq_len:
            t = t[:max_seq_len]
        padded = t + [tokenizer.pad_id] * (max_seq_len - len(t))
        padded_tokens.append(padded)
    input_tokens = torch.tensor(padded_tokens, dtype=torch.long)
    if torch.cuda.is_available():
        input_tokens = input_tokens.cuda()

    # Warmup run
    print("Running warmup inference...")
    try:
        _ = generator.generate(
            input_tokens,
            GeneratorArgs(
                max_gen_len=max_gen_len,
                temperature=temperature,
                top_p=top_p
            )
        )
    except Exception as e:
        print(f"Warmup failed: {e}")
        sys.exit(1)

    # Benchmark run
    print(f"Running benchmark: batch size {batch_size}, max seq len {max_seq_len}, max gen len {max_gen_len}")
    latencies = []
    total_tokens = 0
    num_runs = 10
    for i in range(num_runs):
        start = time.perf_counter()
        try:
            results = generator.generate(
                input_tokens,
                GeneratorArgs(
                    max_gen_len=max_gen_len,
                    temperature=temperature,
                    top_p=top_p
                )
            )
        except Exception as e:
            print(f"Inference run {i} failed: {e}")
            continue
        end = time.perf_counter()
        latency = end - start
        latencies.append(latency)
        # Count generated tokens
        for res in results:
            total_tokens += len(res.generated_tokens)

    if not latencies:
        print("No successful inference runs")
        sys.exit(1)

    # Calculate metrics
    p50_latency = np.percentile(latencies, 50)
    p99_latency = np.percentile(latencies, 99)
    avg_tokens_per_sec = total_tokens / sum(latencies)

    print(f"\nBenchmark Results (batch size {batch_size}):")
    print(f"p50 Latency: {p50_latency:.3f}s")
    print(f"p99 Latency: {p99_latency:.3f}s")
    print(f"Average Throughput: {avg_tokens_per_sec:.0f} tokens/sec")
    print(f"Total Generated Tokens: {total_tokens}")

    return {
        "p50_latency": p50_latency,
        "p99_latency": p99_latency,
        "throughput_tokens_per_sec": avg_tokens_per_sec
    }

if __name__ == "__main__":
    # Example: Run inference with default params
    try:
        metrics = run_llama4_reference_inference(
            model_dir="./llama4_models/8B",
            tokenizer_path="./llama4_models/tokenizer.model",
            prompt="What is the capital of France?",
            batch_size=1,
            max_gen_len=128
        )
    except Exception as e:
        print(f"Fatal error: {e}")
        sys.exit(1)

import json
import time
import torch
import numpy as np
from pathlib import Path
import sys

# Import our earlier TensorRT and Llama 4 inference functions (simplified for example)
# In production, these would be imported from separate modules
try:
    from tensorrt_llama4 import load_tensorrt_engine, run_tensorrt_inference
    from llama4_reference import run_llama4_reference_inference
except ImportError:
    print("Error: Could not import inference modules. Ensure tensorrt_llama4.py and llama4_reference.py are in path.")
    sys.exit(1)

def run_comparison_benchmark(
    model_size: str = "8B",
    precision: str = "fp8",
    batch_sizes: list = [1, 4, 8, 16, 32],
    max_seq_len: int = 2048,
    max_gen_len: int = 128,
    num_runs: int = 10,
    output_file: str = "./benchmark_results.json"
):
    """Runs side-by-side benchmarks of TensorRT and Llama 4 Reference for specified configs.
    Saves results to JSON for reproducibility."""
    # Validate model size
    supported_sizes = ["8B", "70B"]
    if model_size not in supported_sizes:
        raise ValueError(f"Unsupported model size {model_size}. Use {supported_sizes}")

    # Initialize results dict
    results = {
        "metadata": {
            "model_size": model_size,
            "precision": precision,
            "max_seq_len": max_seq_len,
            "max_gen_len": max_gen_len,
            "num_runs": num_runs,
            "hardware": "NVIDIA H100 SXM5 80GB",
            "tensorrt_version": "10.2.0.11",
            "llama4_reference_version": "v2.3.1 (https://github.com/meta/llama)"
        },
        "batch_results": {}
    }

    # Define model paths
    if model_size == "8B":
        trt_engine_path = f"./llama4_8b_{precision}.trt"
        llama_model_dir = "./llama4_models/8B"
    else:
        trt_engine_path = f"./llama4_70b_{precision}.trt"
        llama_model_dir = "./llama4_models/70B"

    # Check if engines/models exist
    if not Path(trt_engine_path).exists():
        print(f"Warning: TensorRT engine {trt_engine_path} not found. Skipping TensorRT benchmarks.")
        trt_available = False
    else:
        trt_available = True
        try:
            trt_engine = load_tensorrt_engine(trt_engine_path)
        except Exception as e:
            print(f"Failed to load TensorRT engine: {e}")
            trt_available = False

    if not Path(llama_model_dir).exists():
        print(f"Warning: Llama 4 model directory {llama_model_dir} not found. Skipping reference benchmarks.")
        llama_available = False
    else:
        llama_available = True

    # Run benchmarks for each batch size
    for bs in batch_sizes:
        print(f"\nBenchmarking batch size {bs}...")
        results["batch_results"][str(bs)] = {}

        # TensorRT benchmark
        if trt_available:
            print(f"Running TensorRT benchmark (batch size {bs})...")
            trt_metrics = {"throughput_tokens_per_sec": 0, "p99_latency_sec": 0, "errors": 0}
            try:
                # Run warmup
                _ = run_tensorrt_inference(trt_engine, batch_size=bs, max_gen_len=max_gen_len, num_runs=3)
                # Run measurement runs
                trt_throughputs = []
                trt_latencies = []
                for _ in range(num_runs):
                    start = time.perf_counter()
                    tokens = run_tensorrt_inference(trt_engine, batch_size=bs, max_gen_len=max_gen_len, num_runs=1)
                    end = time.perf_counter()
                    trt_latencies.append(end - start)
                    trt_throughputs.append(tokens / (end - start))
                trt_metrics["throughput_tokens_per_sec"] = float(np.mean(trt_throughputs))
                trt_metrics["p99_latency_sec"] = float(np.percentile(trt_latencies, 99))
            except Exception as e:
                print(f"TensorRT benchmark failed for batch size {bs}: {e}")
                trt_metrics["errors"] = 1
            results["batch_results"][str(bs)]["tensorrt"] = trt_metrics

        # Llama 4 Reference benchmark
        if llama_available:
            print(f"Running Llama 4 Reference benchmark (batch size {bs})...")
            llama_metrics = {"throughput_tokens_per_sec": 0, "p99_latency_sec": 0, "errors": 0}
            try:
                # We reuse the reference inference function from earlier example
                from llama4_reference import run_llama4_reference_inference
                metrics = run_llama4_reference_inference(
                    model_dir=llama_model_dir,
                    tokenizer_path="./llama4_models/tokenizer.model",
                    prompt="Benchmark prompt for batch size {bs}",
                    batch_size=bs,
                    max_gen_len=max_gen_len,
                    num_runs=num_runs
                )
                llama_metrics["throughput_tokens_per_sec"] = float(metrics["throughput_tokens_per_sec"])
                llama_metrics["p99_latency_sec"] = float(metrics["p99_latency"])
            except Exception as e:
                print(f"Llama 4 Reference benchmark failed for batch size {bs}: {e}")
                llama_metrics["errors"] = 1
            results["batch_results"][str(bs)]["llama4_reference"] = llama_metrics

    # Save results to JSON
    with open(output_file, "w") as f:
        json.dump(results, f, indent=2)
    print(f"\nResults saved to {output_file}")

    # Print summary table
    print("\n" + "="*60)
    print(f"Benchmark Summary: Llama 4 {model_size} {precision.upper()}")
    print("="*60)
    print(f"{'Batch Size':<10} {'TensorRT Throughput':<25} {'Llama 4 Ref Throughput':<25} {'Speedup (TRT)':<10}")
    print("-"*60)
    for bs in batch_sizes:
        bs_str = str(bs)
        if bs_str in results["batch_results"]:
            trt_tp = results["batch_results"][bs_str].get("tensorrt", {}).get("throughput_tokens_per_sec", 0)
            llama_tp = results["batch_results"][bs_str].get("llama4_reference", {}).get("throughput_tokens_per_sec", 0)
            speedup = (trt_tp / llama_tp) if llama_tp > 0 else 0
            print(f"{bs:<10} {trt_tp:<25.0f} {llama_tp:<25.0f} {speedup:<10.2f}x")
    print("="*60)

    return results

if __name__ == "__main__":
    # Run full benchmark for 8B FP8
    try:
        run_comparison_benchmark(
            model_size="8B",
            precision="fp8",
            batch_sizes=[1, 2, 4, 8],
            max_seq_len=2048,
            max_gen_len=128,
            num_runs=10,
            output_file="./llama4_8b_fp8_bench.json"
        )
    except Exception as e:
        print(f"Fatal benchmark error: {e}")
        sys.exit(1)

# Short snippet: Enable FP8 in TensorRT config
config = builder.create_builder_config()
if precision == "fp8":
    config.set_flag(trt.BuilderFlag.FP8)
    config.set_flag(trt.BuilderFlag.OBEY_PRECISION_CONSTRAINTS)
    print("Enabled FP8 quantization for TensorRT engine")

# Short snippet: Load cached TensorRT engine
def load_tensorrt_engine(engine_path: str):
    with open(engine_path, "rb") as f:
        engine_data = f.read()
    runtime = trt.Runtime(TRTLogger())
    return runtime.deserialize_cuda_engine(engine_data)

# Short snippet: Enable Llama 4 reference profiler
generator = Generator(model, tokenizer)
results = generator.generate(
    input_tokens,
    GeneratorArgs(max_gen_len=128),
    profile=True  # Exports trace to llama_profile.json
)