Edge AI Inference Patterns and Architectures for NVIDIA Jetson: A Comprehensive Technical Guide

Published: January 2026 Author: Koca Ventures Technical Team Reading Time: 25 minutes

Plain English Summary

What is Edge AI Inference?

When your AI model makes predictions, that's called "inference." Edge AI inference means running these predictions locally on a device (like a Jetson) instead of sending data to the cloud. It's like having a smart assistant that thinks locally rather than calling a remote server for every question.

Why run AI at the edge?

Key patterns explained simply:

Real-world deployment scenarios:

What will you learn?

1. Triton Inference Server on edge devices
2. Model versioning - managing multiple model versions
3. Federated learning - training without sharing raw data
4. Edge-cloud hybrid - best of both worlds
5. Resource management - power, memory, and thermal limits

The bottom line: Deploying AI at the edge is more than just running a model—it requires smart patterns for updates, testing, and scaling. This guide covers enterprise-ready deployment strategies.

Executive Summary

Edge AI inference on NVIDIA Jetson platforms has evolved dramatically, with the introduction of Jetson Orin, Jetson T4000, and advanced software frameworks enabling sophisticated inference patterns previously exclusive to cloud deployments. This comprehensive guide explores production-ready patterns for deploying, managing, and optimizing AI inference workloads on edge devices, covering everything from Triton Inference Server configurations to federated learning implementations.

The global Edge AI market is projected to reach $66.47 billion by 2030 at a 21.7% CAGR, with manufacturing and healthcare leading adoption. Understanding these patterns is essential for engineers building next-generation edge AI systems.

Table of Contents

1. Triton Inference Server on Edge
2. Model Ensemble and Pipeline Strategies
3. A/B Testing Models on Edge
4. Model Versioning and Hot-Swapping
5. Federated Learning on Edge Devices
6. Online Learning and Model Adaptation
7. Edge-Cloud Hybrid Inference
8. Batching Strategies for Variable Workloads
9. Multi-Model Concurrent Inference
10. Resource Management and Scheduling

1. Triton Inference Server on Edge

Overview

NVIDIA Triton Inference Server provides an optimized inference solution for both cloud and edge deployments. On Jetson platforms, Triton supports TensorFlow, ONNX Runtime, TensorRT, and custom backends, enabling developers to run models directly without conversion while leveraging concurrent model execution, dynamic batching, and model ensembles.

Architecture Diagram

Installation on Jetson

# Download Triton for Jetson from NVIDIA releases
wget https://github.com/triton-inference-server/server/releases/download/v2.61.0/tritonserver2.61.0-agx.tar

# Extract the archive
tar -xvf tritonserver2.61.0-agx.tar

# Set up environment
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/path/to/tritonserver/lib

# Start Triton server
./bin/tritonserver --model-repository=/models --strict-model-config=false

Model Configuration Example

# config.pbtxt for TensorRT model
name: "yolov8_detection"
platform: "tensorrt_plan"
max_batch_size: 8
input [
  {
    name: "images"
    data_type: TYPE_FP32
    dims: [ 3, 640, 640 ]
  }
]
output [
  {
    name: "output0"
    data_type: TYPE_FP32
    dims: [ 84, 8400 ]
  }
]
instance_group [
  {
    count: 2
    kind: KIND_GPU
  }
]
dynamic_batching {
  preferred_batch_size: [ 4, 8 ]
  max_queue_delay_microseconds: 100
}

JetPack-Specific Considerations

On JetPack, while HTTP/REST and gRPC protocols are supported, direct C API integration is recommended for edge use cases to minimize latency overhead. Current limitations include:

• ONNX Runtime backend does not support OpenVino and TensorRT execution providers
• Python backend does not support GPU Tensors and Async BLS
• GPU metrics, GCS/S3/Azure storage are not supported

2. Model Ensemble and Pipeline Strategies

Ensemble Architecture

Model ensembles encapsulate multi-model pipelines (preprocessing → inference → postprocessing) into a single logical unit, eliminating intermediate tensor transfer overhead and minimizing network calls.

Ensemble Configuration

# ensemble_config.pbtxt
name: "vision_pipeline"
platform: "ensemble"
max_batch_size: 8
input [
  {
    name: "RAW_IMAGE"
    data_type: TYPE_UINT8
    dims: [ -1, -1, 3 ]
  }
]
output [
  {
    name: "DETECTIONS"
    data_type: TYPE_FP32
    dims: [ -1, 6 ]
  },
  {
    name: "SEGMENTATION_MASK"
    data_type: TYPE_UINT8
    dims: [ -1, -1 ]
  }
]
ensemble_scheduling {
  step [
    {
      model_name: "image_preprocessor"
      model_version: -1
      input_map {
        key: "raw_input"
        value: "RAW_IMAGE"
      }
      output_map {
        key: "processed_output"
        value: "preprocessed_tensor"
      }
    },
    {
      model_name: "yolov8_detector"
      model_version: -1
      input_map {
        key: "images"
        value: "preprocessed_tensor"
      }
      output_map {
        key: "detections"
        value: "DETECTIONS"
      }
    },
    {
      model_name: "segmentation_model"
      model_version: -1
      input_map {
        key: "input_tensor"
        value: "preprocessed_tensor"
      }
      output_map {
        key: "mask"
        value: "SEGMENTATION_MASK"
      }
    }
  ]
}

Multi-Model Pipeline on Jetson Orin

For Jetson Orin with GPU and DLA, leverage the HaX-CoNN scheduling technique:

# Multi-accelerator pipeline configuration
import tensorrt as trt

class MultiAcceleratorPipeline:
    def __init__(self):
        self.gpu_models = []
        self.dla_models = []

    def configure_dla_model(self, onnx_path, dla_core=0):
        """Configure model to run on DLA accelerator"""
        builder = trt.Builder(trt.Logger(trt.Logger.INFO))
        config = builder.create_builder_config()

        # Enable DLA
        config.default_device_type = trt.DeviceType.DLA
        config.DLA_core = dla_core
        config.set_flag(trt.BuilderFlag.GPU_FALLBACK)
        config.set_flag(trt.BuilderFlag.FP16)

        return self._build_engine(onnx_path, config)

    def configure_gpu_model(self, onnx_path):
        """Configure model to run on GPU"""
        builder = trt.Builder(trt.Logger(trt.Logger.INFO))
        config = builder.create_builder_config()

        config.set_flag(trt.BuilderFlag.FP16)
        config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 1 << 30)

        return self._build_engine(onnx_path, config)

3. A/B Testing Models on Edge

Edge A/B Testing Architecture

Implementation

import hashlib
import time
from dataclasses import dataclass
from typing import Dict, Any, Optional
import sqlite3
import numpy as np

@dataclass
class ABTestConfig:
    experiment_id: str
    model_a_path: str
    model_b_path: str
    traffic_split: float  # Percentage for model A (0.0 to 1.0)
    metrics_db_path: str = "/var/lib/edge-ab/metrics.db"

class EdgeABTestingFramework:
    def __init__(self, config: ABTestConfig):
        self.config = config
        self.model_a = self._load_model(config.model_a_path)
        self.model_b = self._load_model(config.model_b_path)
        self._init_metrics_db()

    def _init_metrics_db(self):
        """Initialize SQLite database for metrics collection"""
        self.conn = sqlite3.connect(self.config.metrics_db_path)
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS ab_metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp REAL,
                experiment_id TEXT,
                model_variant TEXT,
                request_id TEXT,
                latency_ms REAL,
                confidence_score REAL,
                prediction TEXT
            )
        ''')
        self.conn.commit()

    def route_request(self, request_id: str, input_data: np.ndarray) -> Dict[str, Any]:
        """Route request to appropriate model based on consistent hashing"""
        # Use consistent hashing for deterministic routing
        hash_value = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
        normalized = (hash_value % 10000) / 10000.0

        start_time = time.perf_counter()

        if normalized < self.config.traffic_split:
            model_variant = "A"
            result = self.model_a.infer(input_data)
        else:
            model_variant = "B"
            result = self.model_b.infer(input_data)

        latency_ms = (time.perf_counter() - start_time) * 1000

        # Log metrics
        self._log_metrics(
            request_id=request_id,
            model_variant=model_variant,
            latency_ms=latency_ms,
            confidence_score=result.get('confidence', 0.0),
            prediction=str(result.get('prediction', ''))
        )

        return {
            'result': result,
            'model_variant': model_variant,
            'latency_ms': latency_ms
        }

    def _log_metrics(self, **kwargs):
        """Log metrics to local database"""
        self.conn.execute('''
            INSERT INTO ab_metrics
            (timestamp, experiment_id, model_variant, request_id,
             latency_ms, confidence_score, prediction)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            time.time(),
            self.config.experiment_id,
            kwargs['model_variant'],
            kwargs['request_id'],
            kwargs['latency_ms'],
            kwargs['confidence_score'],
            kwargs['prediction']
        ))
        self.conn.commit()

    def get_experiment_summary(self) -> Dict[str, Any]:
        """Generate summary statistics for the experiment"""
        cursor = self.conn.execute('''
            SELECT
                model_variant,
                COUNT(*) as request_count,
                AVG(latency_ms) as avg_latency,
                AVG(confidence_score) as avg_confidence,
                MIN(latency_ms) as min_latency,
                MAX(latency_ms) as max_latency
            FROM ab_metrics
            WHERE experiment_id = ?
            GROUP BY model_variant
        ''', (self.config.experiment_id,))

        return {row[0]: {
            'request_count': row[1],
            'avg_latency_ms': row[2],
            'avg_confidence': row[3],
            'p_min_latency': row[4],
            'p_max_latency': row[5]
        } for row in cursor.fetchall()}

4. Model Versioning and Hot-Swapping

Challenges at the Edge

Unlike cloud deployments where updating a model version is a simple container push, edge environments present unique challenges:

• Devices may connect only intermittently
• Bandwidth constraints limit update frequency
• Geographic distribution across different hardware generations
• Updates may take days to weeks to propagate

Hot-Swapping Architecture

Implementation with Triton Model Control API

import requests
import time
from pathlib import Path
import shutil

class ModelVersionController:
    def __init__(self, triton_url: str, model_repo_path: str):
        self.triton_url = triton_url
        self.model_repo = Path(model_repo_path)

    def deploy_new_version(self, model_name: str, new_version: str,
                           model_file: Path) -> bool:
        """Deploy new model version with zero-downtime hot-swap"""

        version_path = self.model_repo / model_name / new_version
        version_path.mkdir(parents=True, exist_ok=True)

        # Copy model files
        shutil.copy(model_file, version_path / "model.plan")

        # Create version-specific config if needed
        self._create_version_config(model_name, new_version)

        # Trigger Triton to load new version
        response = requests.post(
            f"{self.triton_url}/v2/repository/models/{model_name}/load",
            json={"parameters": {"version": new_version}}
        )

        if response.status_code != 200:
            return False

        # Wait for model to be ready
        return self._wait_for_model_ready(model_name, new_version)

    def switch_active_version(self, model_name: str,
                               from_version: str, to_version: str) -> bool:
        """Atomically switch traffic to new version"""

        # Update model config to set new default version
        config_path = self.model_repo / model_name / "config.pbtxt"

        # Read current config
        with open(config_path, 'r') as f:
            config = f.read()

        # Update version policy
        new_config = self._update_version_policy(config, to_version)

        # Write atomically
        temp_path = config_path.with_suffix('.tmp')
        with open(temp_path, 'w') as f:
            f.write(new_config)
        temp_path.rename(config_path)

        # Reload model
        return self._reload_model(model_name)

    def rollback(self, model_name: str, to_version: str) -> bool:
        """Emergency rollback to previous version"""
        current_version = self._get_current_version(model_name)
        return self.switch_active_version(model_name, current_version, to_version)

    def _wait_for_model_ready(self, model_name: str, version: str,
                               timeout: int = 60) -> bool:
        """Poll until model version is ready"""
        start = time.time()
        while time.time() - start < timeout:
            response = requests.get(
                f"{self.triton_url}/v2/models/{model_name}/versions/{version}/ready"
            )
            if response.status_code == 200:
                return True
            time.sleep(1)
        return False

GitOps-Style Edge Deployment

# edge-deployment-manifest.yaml
apiVersion: edge.ai/v1
kind: ModelDeployment
metadata:
  name: detection-pipeline
  namespace: production
spec:
  models:
    - name: yolov8-detector
      repository: s3://models/yolov8
      version: v2.1.0
      replicas: 2
      resources:
        gpu_memory: 2Gi
        dla_cores: 1
      rollout:
        strategy: canary
        canary_percentage: 10
        success_threshold: 0.95

    - name: classifier
      repository: s3://models/resnet50
      version: v1.3.0
      replicas: 1

  update_policy:
    schedule: "0 2 * * *"  # 2 AM daily
    max_concurrent_updates: 1
    rollback_on_failure: true

  sync:
    interval: 300s  # Pull-based sync every 5 minutes
    source:
      type: git
      url: https://github.com/org/edge-configs.git
      branch: main

5. Federated Learning on Edge Devices

Architecture Overview

Federated learning enables privacy-preserving model training across distributed edge devices without centralizing sensitive data. Recent implementations on Jetson devices demonstrate practical deployment patterns.

<div class="arch-box node">
  <div class="arch-box-header">JETSON NODE #2</div>
  <div class="node-stack">
    <div class="node-item private">Local Data (Private)</div>
    <div class="arch-connector vertical-down small"></div>
    <div class="node-item gpu">Local Train (GPU)</div>
    <div class="arch-connector vertical-down small"></div>
    <div class="node-item encrypted">Gradients (Encrypted)</div>
  </div>
</div>

<div class="arch-box node">
  <div class="arch-box-header">JETSON NODE #3</div>
  <div class="node-stack">
    <div class="node-item private">Local Data (Private)</div>
    <div class="arch-connector vertical-down small"></div>
    <div class="node-item gpu">Local Train (GPU)</div>
    <div class="arch-connector vertical-down small"></div>
    <div class="node-item encrypted">Gradients (Encrypted)</div>
  </div>
</div>

Flower Framework Implementation on Jetson

# flower_client.py - Jetson FL Client
import flwr as fl
import torch
import torch.nn as nn
from torch.utils.data import DataLoader
from typing import Dict, List, Tuple
import numpy as np

class JetsonFlowerClient(fl.client.NumPyClient):
    def __init__(self, model: nn.Module, trainloader: DataLoader,
                 valloader: DataLoader, device: str = "cuda"):
        self.model = model.to(device)
        self.trainloader = trainloader
        self.valloader = valloader
        self.device = device

    def get_parameters(self, config) -> List[np.ndarray]:
        """Return model parameters as numpy arrays"""
        return [val.cpu().numpy() for _, val in self.model.state_dict().items()]

    def set_parameters(self, parameters: List[np.ndarray]) -> None:
        """Set model parameters from numpy arrays"""
        params_dict = zip(self.model.state_dict().keys(), parameters)
        state_dict = {k: torch.tensor(v) for k, v in params_dict}
        self.model.load_state_dict(state_dict, strict=True)

    def fit(self, parameters: List[np.ndarray],
            config: Dict) -> Tuple[List[np.ndarray], int, Dict]:
        """Train model on local data"""
        self.set_parameters(parameters)

        epochs = config.get("local_epochs", 1)
        lr = config.get("learning_rate", 0.001)

        optimizer = torch.optim.Adam(self.model.parameters(), lr=lr)
        criterion = nn.CrossEntropyLoss()

        self.model.train()
        for epoch in range(epochs):
            for batch_idx, (data, target) in enumerate(self.trainloader):
                data, target = data.to(self.device), target.to(self.device)
                optimizer.zero_grad()
                output = self.model(data)
                loss = criterion(output, target)
                loss.backward()
                optimizer.step()

        return self.get_parameters(config), len(self.trainloader.dataset), {}

    def evaluate(self, parameters: List[np.ndarray],
                 config: Dict) -> Tuple[float, int, Dict]:
        """Evaluate model on local validation data"""
        self.set_parameters(parameters)

        self.model.eval()
        loss = 0.0
        correct = 0
        criterion = nn.CrossEntropyLoss()

        with torch.no_grad():
            for data, target in self.valloader:
                data, target = data.to(self.device), target.to(self.device)
                output = self.model(data)
                loss += criterion(output, target).item()
                pred = output.argmax(dim=1, keepdim=True)
                correct += pred.eq(target.view_as(pred)).sum().item()

        accuracy = correct / len(self.valloader.dataset)
        return float(loss), len(self.valloader.dataset), {"accuracy": accuracy}

# Start Flower client
def main():
    model = create_model()  # Your model architecture
    trainloader, valloader = load_local_data()

    client = JetsonFlowerClient(model, trainloader, valloader)

    fl.client.start_numpy_client(
        server_address="aggregator.local:8080",
        client=client,
        grpc_max_message_length=1024*1024*1024  # 1GB for large models
    )

if __name__ == "__main__":
    main()

Federated Aggregation Server

# fed_server.py - Aggregation Server
import flwr as fl
from flwr.server.strategy import FedAvg
from typing import List, Tuple, Dict, Optional
import numpy as np

class EdgeFederatedStrategy(FedAvg):
    def __init__(self,
                 min_fit_clients: int = 3,
                 min_available_clients: int = 3,
                 energy_budget: float = 100.0):  # Joules
        super().__init__(
            min_fit_clients=min_fit_clients,
            min_available_clients=min_available_clients,
        )
        self.energy_budget = energy_budget
        self.client_energy_usage = {}

    def configure_fit(self, server_round: int,
                      parameters, client_manager) -> List[Tuple]:
        """Configure training round with energy-aware client selection"""

        # Sample clients based on energy budget
        sample_size = min(self.min_fit_clients,
                         client_manager.num_available())

        clients = client_manager.sample(
            num_clients=sample_size,
            min_num_clients=self.min_fit_clients
        )

        # Configure based on device capabilities
        config = {
            "local_epochs": 5 if server_round < 10 else 3,
            "learning_rate": 0.001 * (0.95 ** server_round),
            "batch_size": 32
        }

        return [(client, fl.common.FitIns(parameters, config))
                for client in clients]

def start_server():
    strategy = EdgeFederatedStrategy(
        min_fit_clients=3,
        min_available_clients=5,
        energy_budget=50.0
    )

    fl.server.start_server(
        server_address="0.0.0.0:8080",
        config=fl.server.ServerConfig(num_rounds=100),
        strategy=strategy
    )

if __name__ == "__main__":
    start_server()

Performance Benchmarks

Based on research with Jetson devices running federated learning:

6. Online Learning and Model Adaptation

MicroAdapt: Self-Evolving Edge AI

Recent research from Osaka University's SANKEN has developed MicroAdapt, a self-evolving edge AI technology achieving:

• 100,000x faster processing compared to conventional deep learning
• 60% higher accuracy through continuous adaptation
• Real-time learning and forecasting on compact devices

Continual Learning Implementation

import torch
import torch.nn as nn
from collections import deque
import numpy as np

class ContinualLearningEdge:
    def __init__(self, base_model: nn.Module,
                 buffer_size: int = 1000,
                 update_frequency: int = 100):
        self.model = base_model
        self.replay_buffer = deque(maxlen=buffer_size)
        self.update_frequency = update_frequency
        self.sample_count = 0
        self.ewc_lambda = 0.5  # Elastic Weight Consolidation
        self.fisher_information = {}
        self.previous_params = {}

    def infer_and_adapt(self, input_data: torch.Tensor,
                        ground_truth: Optional[torch.Tensor] = None):
        """Perform inference and optionally adapt model"""

        # Forward pass
        with torch.no_grad():
            prediction = self.model(input_data)

        # If ground truth available, store for replay
        if ground_truth is not None:
            self.replay_buffer.append((input_data.clone(), ground_truth.clone()))
            self.sample_count += 1

            # Trigger adaptation
            if self.sample_count % self.update_frequency == 0:
                self._adapt_model()

        return prediction

    def _adapt_model(self):
        """Adapt model using replay buffer with EWC regularization"""
        if len(self.replay_buffer) < 32:
            return

        self.model.train()
        optimizer = torch.optim.Adam(self.model.parameters(), lr=0.0001)
        criterion = nn.CrossEntropyLoss()

        # Sample from replay buffer
        indices = np.random.choice(len(self.replay_buffer),
                                   min(32, len(self.replay_buffer)),
                                   replace=False)

        batch_inputs = torch.stack([self.replay_buffer[i][0] for i in indices])
        batch_targets = torch.stack([self.replay_buffer[i][1] for i in indices])

        optimizer.zero_grad()
        outputs = self.model(batch_inputs)

        # Task loss
        task_loss = criterion(outputs, batch_targets)

        # EWC regularization loss
        ewc_loss = self._compute_ewc_loss()

        total_loss = task_loss + self.ewc_lambda * ewc_loss
        total_loss.backward()
        optimizer.step()

        self.model.eval()

    def _compute_ewc_loss(self) -> torch.Tensor:
        """Compute Elastic Weight Consolidation loss"""
        ewc_loss = 0.0
        for name, param in self.model.named_parameters():
            if name in self.fisher_information:
                fisher = self.fisher_information[name]
                prev_param = self.previous_params[name]
                ewc_loss += (fisher * (param - prev_param) ** 2).sum()
        return ewc_loss

    def consolidate_knowledge(self):
        """Update Fisher information matrix after task completion"""
        self.model.eval()

        for name, param in self.model.named_parameters():
            self.previous_params[name] = param.clone().detach()

            # Compute Fisher information
            self.fisher_information[name] = torch.zeros_like(param)

        # Approximate Fisher using replay buffer
        for input_data, target in self.replay_buffer:
            self.model.zero_grad()
            output = self.model(input_data.unsqueeze(0))
            loss = nn.functional.cross_entropy(output, target.unsqueeze(0))
            loss.backward()

            for name, param in self.model.named_parameters():
                if param.grad is not None:
                    self.fisher_information[name] += param.grad ** 2

        # Normalize
        for name in self.fisher_information:
            self.fisher_information[name] /= len(self.replay_buffer)

7. Edge-Cloud Hybrid Inference

Hybrid Architecture Benefits

Research demonstrates that hybrid edge-cloud architectures achieve:

• 46% lower P99 latency compared to monolithic approaches
• 67% higher throughput with 10,000 concurrent users
• 99.5% bandwidth reduction through selective cloud offloading

  <div class="arch-row centered">
    <div class="arch-box classifier">
      <div class="arch-box-header">Request Classifier</div>
      <div class="arch-box-content">(Complexity Estimator)</div>
    </div>
    <div class="path-label low-latency">&#8592; Low Latency Path</div>
  </div>

  <div class="arch-connector split-down two-way"></div>

  <div class="arch-row decision">
    <div class="arch-box local">
      <div class="arch-box-header">Simple Tasks</div>
      <div class="arch-box-content">(Local)</div>
      <div class="arch-connector vertical-down small"></div>
      <div class="engine-box">TensorRT Engine</div>
    </div>
    <div class="arch-box offload">
      <div class="arch-box-header">Complex Tasks</div>
      <div class="arch-box-content">(Offload)</div>
    </div>
  </div>
</div>

import numpy as np
from typing import Tuple, Dict, Any
from dataclasses import dataclass
import time

@dataclass
class RoutingDecision:
    route_to_edge: bool
    confidence: float
    estimated_latency_ms: float
    estimated_accuracy: float

class HybridInferenceRouter:
    def __init__(self,
                 edge_model_capacity: float = 0.7,
                 latency_threshold_ms: float = 100.0,
                 cloud_endpoint: str = "https://cloud-inference.api/v1"):
        self.edge_capacity = edge_model_capacity
        self.latency_threshold = latency_threshold_ms
        self.cloud_endpoint = cloud_endpoint

        # Performance tracking
        self.edge_latencies = []
        self.cloud_latencies = []
        self.edge_accuracies = []

    def route_request(self, input_features: np.ndarray,
                      complexity_score: float) -> RoutingDecision:
        """Determine whether to process locally or offload to cloud"""

        # Estimate edge processing capability
        edge_confidence = self._estimate_edge_confidence(
            input_features, complexity_score
        )

        # Check current edge load
        edge_available = self._check_edge_availability()

        # Calculate expected latencies
        edge_latency = self._estimate_edge_latency(complexity_score)
        cloud_latency = self._estimate_cloud_latency()

        # Decision logic
        if complexity_score < self.edge_capacity and edge_available:
            if edge_latency < self.latency_threshold:
                return RoutingDecision(
                    route_to_edge=True,
                    confidence=edge_confidence,
                    estimated_latency_ms=edge_latency,
                    estimated_accuracy=self._get_edge_accuracy_estimate()
                )

        return RoutingDecision(
            route_to_edge=False,
            confidence=1.0 - edge_confidence,
            estimated_latency_ms=cloud_latency,
            estimated_accuracy=0.95  # Cloud typically higher accuracy
        )

    async def infer(self, input_data: np.ndarray) -> Dict[str, Any]:
        """Execute inference with automatic routing"""

        complexity = self._compute_complexity(input_data)
        decision = self.route_request(input_data, complexity)

        start_time = time.perf_counter()

        if decision.route_to_edge:
            result = await self._edge_inference(input_data)
            latency = (time.perf_counter() - start_time) * 1000
            self.edge_latencies.append(latency)
        else:
            result = await self._cloud_inference(input_data)
            latency = (time.perf_counter() - start_time) * 1000
            self.cloud_latencies.append(latency)

        return {
            'result': result,
            'routed_to': 'edge' if decision.route_to_edge else 'cloud',
            'latency_ms': latency,
            'decision_confidence': decision.confidence
        }

    def _compute_complexity(self, input_data: np.ndarray) -> float:
        """Estimate input complexity for routing decision"""
        # Simple heuristic: variance and size-based complexity
        variance = np.var(input_data)
        size_factor = np.prod(input_data.shape) / 1000000  # Normalize
        return min(1.0, (variance * 0.5 + size_factor * 0.5))

# config.pbtxt with advanced batching
name: "detection_model"
platform: "tensorrt_plan"
max_batch_size: 32

dynamic_batching {
    # Preferred batch sizes for optimal GPU utilization
    preferred_batch_size: [ 4, 8, 16, 32 ]

    # Maximum queue delay before forcing batch execution
    max_queue_delay_microseconds: 5000

    # Priority levels for request scheduling
    priority_levels: 3
    default_priority_level: 1

    # Queue policies per priority
    priority_queue_policy {
        key: 1
        value: {
            timeout_action: DELAY
            default_timeout_microseconds: 10000
            allow_timeout_override: true
            max_queue_size: 100
        }
    }
    priority_queue_policy {
        key: 2
        value: {
            timeout_action: REJECT
            default_timeout_microseconds: 5000
            max_queue_size: 50
        }
    }
    priority_queue_policy {
        key: 3
        value: {
            timeout_action: REJECT
            default_timeout_microseconds: 1000
            max_queue_size: 10
        }
    }
}

# Sequence batching for stateful models
sequence_batching {
    max_sequence_idle_microseconds: 5000000
    control_input [
        {
            name: "START"
            control [
                {
                    kind: CONTROL_SEQUENCE_START
                    fp32_false_true: [ 0, 1 ]
                }
            ]
        },
        {
            name: "END"
            control [
                {
                    kind: CONTROL_SEQUENCE_END
                    fp32_false_true: [ 0, 1 ]
                }
            ]
        }
    ]
}

import threading
import time
from collections import deque
from typing import Callable, Any
import numpy as np

class AdaptiveBatchController:
    def __init__(self,
                 min_batch_size: int = 1,
                 max_batch_size: int = 32,
                 target_latency_ms: float = 50.0,
                 adjustment_interval: float = 1.0):
        self.min_batch = min_batch_size
        self.max_batch = max_batch_size
        self.target_latency = target_latency_ms
        self.adjustment_interval = adjustment_interval

        self.current_batch_size = min_batch_size
        self.latency_history = deque(maxlen=100)
        self.throughput_history = deque(maxlen=100)

        self._lock = threading.Lock()
        self._running = False

    def start_adaptive_control(self):
        """Start background thread for batch size adjustment"""
        self._running = True
        self._control_thread = threading.Thread(target=self._control_loop)
        self._control_thread.start()

    def stop(self):
        self._running = False
        self._control_thread.join()

    def _control_loop(self):
        """PID-like controller for batch size"""
        while self._running:
            time.sleep(self.adjustment_interval)

            with self._lock:
                if len(self.latency_history) < 10:
                    continue

                avg_latency = np.mean(list(self.latency_history)[-20:])
                latency_trend = self._compute_trend(self.latency_history)

                # Adjust batch size based on latency
                if avg_latency < self.target_latency * 0.8:
                    # Under target, can increase batch size
                    self.current_batch_size = min(
                        self.max_batch,
                        self.current_batch_size + 2
                    )
                elif avg_latency > self.target_latency * 1.2:
                    # Over target, decrease batch size
                    self.current_batch_size = max(
                        self.min_batch,
                        self.current_batch_size - 2
                    )
                elif latency_trend > 0.1:  # Latency trending up
                    self.current_batch_size = max(
                        self.min_batch,
                        self.current_batch_size - 1
                    )

    def record_inference(self, batch_size: int, latency_ms: float):
        """Record inference metrics"""
        with self._lock:
            self.latency_history.append(latency_ms)
            self.throughput_history.append(batch_size / (latency_ms / 1000))

    def get_recommended_batch_size(self) -> int:
        """Get current recommended batch size"""
        with self._lock:
            return self.current_batch_size

    def _compute_trend(self, history: deque) -> float:
        """Compute trend using linear regression slope"""
        if len(history) < 5:
            return 0.0
        recent = list(history)[-20:]
        x = np.arange(len(recent))
        slope = np.polyfit(x, recent, 1)[0]
        return slope / np.mean(recent)  # Normalized slope

  <div class="arch-row">
    <div class="arch-box">
      <div class="arch-box-header">CUDA MPS</div>
      <div class="arch-box-content">(Multi-Process Service)</div>
    </div>
    <div class="arch-box">
      <div class="arch-box-header">Stream Manager</div>
      <div class="arch-box-content">(Priority-Aware)</div>
    </div>
  </div>

  <div class="arch-connector merge-down"></div>

  <div class="arch-box context-pool">
    <div class="arch-box-header">CUDA CONTEXT POOL</div>
    <div class="stream-grid">
      <div class="stream-column">
        <div class="stream-header">Stream 0<br><span class="priority">(Priority 0)</span></div>
        <div class="arch-connector vertical-down small"></div>
        <div class="model-box primary">YOLO v8<br>Detect</div>
      </div>
      <div class="stream-column">
        <div class="stream-header">Stream 1<br><span class="priority">(Priority 1)</span></div>
        <div class="arch-connector vertical-down small"></div>
        <div class="model-box">ResNet<br>Classif</div>
      </div>
      <div class="stream-column">
        <div class="stream-header">Stream 2<br><span class="priority">(Priority 1)</span></div>
        <div class="arch-connector vertical-down small"></div>
        <div class="model-box">SegNet<br>Segment</div>
      </div>
      <div class="stream-column">
        <div class="stream-header">Stream 3<br><span class="priority">(Priority 2)</span></div>
        <div class="arch-connector vertical-down small"></div>
        <div class="model-box">Custom<br>Model</div>
      </div>
    </div>
  </div>
</div>

  <div class="gpu-sm-container">
    <div class="sm-label">GPU SMs</div>
    <div class="sm-grid">
      <span class="sm-unit">SM0</span>
      <span class="sm-unit">SM1</span>
      <span class="sm-unit">SM2</span>
      <span class="sm-unit">SM3</span>
      <span class="sm-unit">SM4</span>
      <span class="sm-unit">SM5</span>
      <span class="sm-unit">SM6</span>
      <span class="sm-unit">SM7</span>
    </div>
    <div class="sm-shared-label">Shared by all streams</div>
  </div>

  <div class="dla-container">
    <div class="arch-box dla">
      <div class="arch-box-header">DLA 0</div>
      <div class="arch-box-content">Dedicated Model</div>
    </div>
    <div class="arch-box dla">
      <div class="arch-box-header">DLA 1</div>
      <div class="arch-box-content">Dedicated Model</div>
    </div>
    <span class="dla-note">(Xavier/Orin only)</span>
  </div>
</div>

#!/bin/bash
# enable_mps.sh - Enable CUDA MPS on Jetson

# Set exclusive compute mode
sudo nvidia-smi -i 0 -c EXCLUSIVE_PROCESS

# Start MPS control daemon
sudo nvidia-cuda-mps-control -d

# Set active thread percentage (optional, for resource limiting)
echo "set_default_active_thread_percentage 50" | sudo nvidia-cuda-mps-control

# Verify MPS is running
echo "get_server_list" | nvidia-cuda-mps-control

import asyncio
import torch
import torch.cuda as cuda
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import queue

@dataclass
class InferenceTask:
    model_name: str
    input_data: torch.Tensor
    priority: int = 1
    callback: Optional[callable] = None

class MultiModelInferenceManager:
    def __init__(self, model_configs: Dict[str, Dict]):
        self.models = {}
        self.streams = {}
        self.executors = {}

        # Initialize CUDA streams with priorities
        for name, config in model_configs.items():
            priority = config.get('priority', 0)
            # CUDA stream priorities: lower number = higher priority
            # Range is typically [-1, 0] on consumer GPUs
            stream_priority = max(-1, min(0, -priority))

            self.streams[name] = cuda.Stream(priority=stream_priority)
            self.models[name] = self._load_model(config['path'])

        self.task_queue = asyncio.PriorityQueue()
        self._running = False

    def _load_model(self, path: str) -> torch.nn.Module:
        """Load TensorRT or PyTorch model"""
        if path.endswith('.plan'):
            return self._load_tensorrt_engine(path)
        else:
            model = torch.jit.load(path)
            model.cuda()
            model.eval()
            return model

    async def submit_inference(self, task: InferenceTask) -> asyncio.Future:
        """Submit inference task to queue"""
        future = asyncio.Future()
        await self.task_queue.put((task.priority, task, future))
        return future

    async def process_queue(self):
        """Process inference tasks from queue"""
        self._running = True

        while self._running:
            try:
                priority, task, future = await asyncio.wait_for(
                    self.task_queue.get(), timeout=0.1
                )
            except asyncio.TimeoutError:
                continue

            # Execute inference on appropriate stream
            stream = self.streams[task.model_name]
            model = self.models[task.model_name]

            try:
                with cuda.stream(stream):
                    input_gpu = task.input_data.cuda(non_blocking=True)
                    with torch.no_grad():
                        output = model(input_gpu)

                    # Synchronize stream
                    stream.synchronize()

                    result = output.cpu()
                    future.set_result(result)

                    if task.callback:
                        task.callback(result)

            except Exception as e:
                future.set_exception(e)

    async def batch_inference(self, tasks: List[InferenceTask]) -> List[Any]:
        """Execute multiple inference tasks concurrently"""
        futures = []

        for task in tasks:
            future = await self.submit_inference(task)
            futures.append(future)

        results = await asyncio.gather(*futures, return_exceptions=True)
        return results

    def get_utilization_stats(self) -> Dict[str, float]:
        """Get GPU utilization statistics per model"""
        stats = {}
        for name, stream in self.streams.items():
            # Query stream status
            stats[name] = {
                'pending_operations': stream.query(),
                'priority': stream.priority
            }
        return stats

import subprocess
import json
from typing import Dict, Any
from dataclasses import dataclass
import time

@dataclass
class JetsonPowerProfile:
    name: str
    gpu_freq_mhz: int
    cpu_freq_mhz: int
    dla_freq_mhz: int
    power_budget_watts: float

class JetsonResourceManager:
    POWER_PROFILES = {
        'max_performance': JetsonPowerProfile(
            name='MAXN',
            gpu_freq_mhz=1300,
            cpu_freq_mhz=2200,
            dla_freq_mhz=1600,
            power_budget_watts=60.0
        ),
        'balanced': JetsonPowerProfile(
            name='30W',
            gpu_freq_mhz=900,
            cpu_freq_mhz=1500,
            dla_freq_mhz=1100,
            power_budget_watts=30.0
        ),
        'power_save': JetsonPowerProfile(
            name='15W',
            gpu_freq_mhz=600,
            cpu_freq_mhz=1000,
            dla_freq_mhz=800,
            power_budget_watts=15.0
        )
    }

    def __init__(self):
        self.current_profile = None
        self._thermal_threshold = 80.0  # Celsius

    def set_power_mode(self, profile_name: str) -> bool:
        """Set Jetson power mode"""
        if profile_name not in self.POWER_PROFILES:
            return False

        profile = self.POWER_PROFILES[profile_name]

        try:
            # Use nvpmodel to set power mode
            subprocess.run(
                ['sudo', 'nvpmodel', '-m', profile.name],
                check=True, capture_output=True
            )

            # Set Jetson clocks for maximum performance
            if profile_name == 'max_performance':
                subprocess.run(
                    ['sudo', 'jetson_clocks'],
                    check=True, capture_output=True
                )

            self.current_profile = profile
            return True

        except subprocess.CalledProcessError:
            return False

    def get_system_stats(self) -> Dict[str, Any]:
        """Get current system statistics"""
        stats = {}

        # Read GPU utilization
        try:
            with open('/sys/devices/gpu.0/load', 'r') as f:
                stats['gpu_utilization'] = int(f.read().strip()) / 10.0
        except FileNotFoundError:
            stats['gpu_utilization'] = 0.0

        # Read temperatures
        temps = {}
        temp_zones = [
            '/sys/devices/virtual/thermal/thermal_zone0/temp',
            '/sys/devices/virtual/thermal/thermal_zone1/temp',
            '/sys/devices/virtual/thermal/thermal_zone2/temp'
        ]
        for i, zone in enumerate(temp_zones):
            try:
                with open(zone, 'r') as f:
                    temps[f'zone_{i}'] = int(f.read().strip()) / 1000.0
            except FileNotFoundError:
                pass
        stats['temperatures'] = temps

        # Read power consumption
        try:
            result = subprocess.run(
                ['tegrastats', '--interval', '100'],
                capture_output=True, timeout=0.5
            )
            # Parse tegrastats output for power info
            stats['power_mw'] = self._parse_tegrastats(result.stdout.decode())
        except:
            stats['power_mw'] = 0

        return stats

    def adaptive_throttling(self, target_fps: float,
                            current_fps: float) -> None:
        """Adjust power profile based on performance targets"""
        stats = self.get_system_stats()
        max_temp = max(stats['temperatures'].values()) if stats['temperatures'] else 0

        # Thermal throttling
        if max_temp > self._thermal_threshold:
            if self.current_profile != self.POWER_PROFILES['power_save']:
                self.set_power_mode('power_save')
            return

        # Performance-based adjustment
        fps_ratio = current_fps / target_fps

        if fps_ratio < 0.8:  # Under-performing
            self.set_power_mode('max_performance')
        elif fps_ratio > 1.2 and max_temp < 70:  # Over-performing, room for power save
            self.set_power_mode('balanced')

# k3s-triton-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: triton-inference-server
  namespace: edge-ai
spec:
  replicas: 1
  selector:
    matchLabels:
      app: triton-server
  template:
    metadata:
      labels:
        app: triton-server
    spec:
      runtimeClassName: nvidia
      containers:
      - name: triton
        image: nvcr.io/nvidia/tritonserver:24.01-py3-jetson
        ports:
        - containerPort: 8000
          name: http
        - containerPort: 8001
          name: grpc
        - containerPort: 8002
          name: metrics
        resources:
          limits:
            nvidia.com/gpu: 1
            memory: "8Gi"
          requests:
            memory: "4Gi"
        volumeMounts:
        - name: model-repository
          mountPath: /models
        - name: shm
          mountPath: /dev/shm
        args:
        - tritonserver
        - --model-repository=/models
        - --strict-model-config=false
        - --log-verbose=1
        livenessProbe:
          httpGet:
            path: /v2/health/live
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /v2/health/ready
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
      volumes:
      - name: model-repository
        persistentVolumeClaim:
          claimName: model-pvc
      - name: shm
        emptyDir:
          medium: Memory
          sizeLimit: "2Gi"
---
apiVersion: v1
kind: Service
metadata:
  name: triton-service
  namespace: edge-ai
spec:
  selector:
    app: triton-server
  ports:
  - name: http
    port: 8000
    targetPort: 8000
  - name: grpc
    port: 8001
    targetPort: 8001
  - name: metrics
    port: 8002
    targetPort: 8002
  type: LoadBalancer