INTRODUCTION
Modern large language model applications face significant challenges in deployment across heterogeneous hardware environments. Organizations and individual developers need systems that can automatically detect available GPU hardware, select optimal execution backends, manage multiple model sources both local and remote, implement failover mechanisms when primary models become unavailable, and intelligently route user prompts to the most appropriate model based on task characteristics. This article presents a comprehensive architecture for building production-ready LLM applications that address all these requirements.
The core challenge lies in creating a unified abstraction layer that handles the complexity of different GPU vendors including Nvidia CUDA, AMD ROCm, Intel oneAPI, and Apple Metal Performance Shaders while providing seamless model switching capabilities. Additionally, the system must make intelligent decisions about which model to use for a given task, considering factors such as model capabilities, computational requirements, latency constraints, and availability.
GPU HARDWARE DETECTION AND SELECTION
The first critical component of an intelligent LLM application is the ability to detect and utilize available GPU hardware. Different GPU vendors provide different software stacks and APIs. Nvidia uses CUDA, AMD provides ROCm for their GPUs, Intel offers oneAPI and Level Zero, and Apple provides Metal Performance Shaders for their unified memory architecture. A robust detection system must probe for all these backends and select the optimal one based on performance characteristics.
The detection process begins by attempting to import and initialize each GPU backend library. For Nvidia GPUs, we check for CUDA availability through PyTorch or TensorFlow. For AMD, we look for ROCm support. Intel GPUs are detected through oneAPI or OpenCL interfaces. Apple Silicon detection happens through the Metal framework availability check. Each detection attempt must be wrapped in exception handling because the libraries may not be installed or the hardware may not be present.
Here is a foundational GPU detection implementation:
import sys
import subprocess
from typing import Optional, Dict, List, Tuple
from enum import Enum
class GPUBackend(Enum):
CUDA = "cuda"
ROCM = "rocm"
MPS = "mps"
INTEL = "intel"
CPU = "cpu"
class GPUDetector:
def __init__(self):
self.available_backends = []
self.backend_info = {}
self.detect_all_backends()
def detect_cuda(self) -> Tuple[bool, Dict]:
"""Detect Nvidia CUDA GPUs and gather information"""
try:
import torch
if torch.cuda.is_available():
device_count = torch.cuda.device_count()
devices = []
for i in range(device_count):
props = torch.cuda.get_device_properties(i)
devices.append({
'index': i,
'name': props.name,
'compute_capability': f"{props.major}.{props.minor}",
'total_memory': props.total_memory,
'multi_processor_count': props.multi_processor_count
})
return True, {'device_count': device_count, 'devices': devices}
except Exception as e:
pass
return False, {}
The CUDA detection function attempts to import PyTorch and query CUDA availability. If successful, it enumerates all available CUDA devices and collects detailed information including device name, compute capability, total memory, and multiprocessor count. This information is crucial for later performance optimization decisions. The compute capability version determines which CUDA features are available and influences model loading strategies.
For AMD ROCm detection, the process is similar but uses ROCm-specific APIs:
def detect_rocm(self) -> Tuple[bool, Dict]:
"""Detect AMD ROCm GPUs"""
try:
import torch
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
if torch.cuda.is_available():
device_count = torch.cuda.device_count()
devices = []
for i in range(device_count):
props = torch.cuda.get_device_properties(i)
devices.append({
'index': i,
'name': props.name,
'total_memory': props.total_memory,
'gcn_arch': props.gcnArchName if hasattr(props, 'gcnArchName') else 'unknown'
})
return True, {'device_count': device_count, 'devices': devices}
except Exception as e:
pass
try:
result = subprocess.run(['rocm-smi', '--showproductname'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0 and result.stdout:
return True, {'detected_via': 'rocm-smi', 'info': result.stdout.strip()}
except Exception as e:
pass
return False, {}
ROCm detection is more complex because PyTorch built with ROCm support uses a CUDA-compatible API but runs on AMD hardware. We first check for the HIP version in PyTorch which indicates ROCm support. Additionally, we attempt to run the rocm-smi command-line tool which provides system management information for AMD GPUs. This dual approach ensures detection even when PyTorch is not available or not built with ROCm support.
Apple Metal Performance Shaders detection for Apple Silicon:
def detect_mps(self) -> Tuple[bool, Dict]:
"""Detect Apple Metal Performance Shaders"""
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return True, {'backend': 'mps', 'built': torch.backends.mps.is_built()}
except Exception as e:
pass
try:
import platform
if platform.system() == 'Darwin' and platform.machine() == 'arm64':
return True, {'platform': 'Apple Silicon', 'detected_via': 'platform'}
except Exception as e:
pass
return False, {}
Apple Silicon detection checks for MPS availability in PyTorch backends. Since MPS is only available on Apple Silicon Macs, we also verify the platform is Darwin with ARM64 architecture. This provides a fallback detection method when PyTorch is not available but the application is running on compatible hardware.
Intel GPU detection:
def detect_intel(self) -> Tuple[bool, Dict]:
"""Detect Intel GPUs with oneAPI support"""
try:
import torch
if hasattr(torch, 'xpu') and torch.xpu.is_available():
device_count = torch.xpu.device_count()
devices = []
for i in range(device_count):
devices.append({
'index': i,
'name': torch.xpu.get_device_name(i)
})
return True, {'device_count': device_count, 'devices': devices}
except Exception as e:
pass
try:
result = subprocess.run(['sycl-ls'], capture_output=True, text=True, timeout=5)
if result.returncode == 0 and 'gpu' in result.stdout.lower():
return True, {'detected_via': 'sycl-ls', 'info': result.stdout.strip()}
except Exception as e:
pass
return False, {}
Intel GPU detection looks for Intel Extension for PyTorch XPU support. The XPU device represents Intel GPUs accessible through oneAPI. We also attempt to run sycl-ls which lists all SYCL devices including Intel GPUs. This command-line tool is part of the Intel oneAPI toolkit.
After detecting all available backends, the system must select the optimal one. The selection process considers multiple factors including raw computational power, memory bandwidth, driver stability, and framework support quality. Generally, the priority order is CUDA for Nvidia GPUs due to mature ecosystem support, followed by ROCm for AMD, MPS for Apple Silicon, Intel for Intel GPUs, and CPU as the fallback.
def detect_all_backends(self):
"""Detect all available GPU backends and rank them"""
cuda_available, cuda_info = self.detect_cuda()
if cuda_available:
self.available_backends.append(GPUBackend.CUDA)
self.backend_info[GPUBackend.CUDA] = cuda_info
rocm_available, rocm_info = self.detect_rocm()
if rocm_available and not cuda_available:
self.available_backends.append(GPUBackend.ROCM)
self.backend_info[GPUBackend.ROCM] = rocm_info
mps_available, mps_info = self.detect_mps()
if mps_available:
self.available_backends.append(GPUBackend.MPS)
self.backend_info[GPUBackend.MPS] = mps_info
intel_available, intel_info = self.detect_intel()
if intel_available:
self.available_backends.append(GPUBackend.INTEL)
self.backend_info[GPUBackend.INTEL] = intel_info
self.available_backends.append(GPUBackend.CPU)
self.backend_info[GPUBackend.CPU] = {'fallback': True}
def get_optimal_backend(self) -> GPUBackend:
"""Select the optimal backend based on availability and priority"""
priority_order = [GPUBackend.CUDA, GPUBackend.ROCM, GPUBackend.MPS,
GPUBackend.INTEL, GPUBackend.CPU]
for backend in priority_order:
if backend in self.available_backends:
return backend
return GPUBackend.CPU
The detection orchestration method probes all backends systematically. Note that ROCm detection only registers if CUDA is not available, preventing double-counting of AMD GPUs that might appear through both interfaces. The optimal backend selection follows the priority order, returning the first available backend from the prioritized list.
MODEL LOADING AND MANAGEMENT ARCHITECTURE
Once the optimal GPU backend is determined, the next challenge is loading and managing LLM models from various sources. Models may reside locally on disk, be downloaded from remote repositories like Hugging Face, or be accessed through API endpoints. The model management system must provide a unified interface for all these sources while handling caching, version management, and resource allocation.
The model manager maintains a registry of available models with metadata including model type, source location, size, quantization level, and hardware requirements. This registry enables intelligent model selection based on task requirements and available resources.
from dataclasses import dataclass
from typing import Optional, Union, Any
from pathlib import Path
import json
import hashlib
@dataclass
class ModelConfig:
model_id: str
model_type: str
source: str
local_path: Optional[Path]
remote_url: Optional[str]
api_endpoint: Optional[str]
size_gb: float
quantization: Optional[str]
context_length: int
required_memory_gb: float
supported_backends: List[GPUBackend]
capabilities: List[str]
priority: int
class ModelRegistry:
def __init__(self, registry_path: Optional[Path] = None):
self.registry_path = registry_path or Path.home() / '.llm_app' / 'models.json'
self.models = {}
self.load_registry()
def load_registry(self):
"""Load model registry from disk"""
if self.registry_path.exists():
with open(self.registry_path, 'r') as f:
data = json.load(f)
for model_id, config_dict in data.items():
config_dict['local_path'] = Path(config_dict['local_path']) if config_dict.get('local_path') else None
config_dict['supported_backends'] = [GPUBackend(b) for b in config_dict['supported_backends']]
self.models[model_id] = ModelConfig(**config_dict)
else:
self.initialize_default_registry()
def save_registry(self):
"""Persist model registry to disk"""
self.registry_path.parent.mkdir(parents=True, exist_ok=True)
data = {}
for model_id, config in self.models.items():
config_dict = config.__dict__.copy()
config_dict['local_path'] = str(config_dict['local_path']) if config_dict['local_path'] else None
config_dict['supported_backends'] = [b.value for b in config_dict['supported_backends']]
data[model_id] = config_dict
with open(self.registry_path, 'w') as f:
json.dump(data, f, indent=2)
The ModelConfig dataclass encapsulates all relevant information about a model. The source field indicates whether the model is local, remote, or API-based. The local_path points to the model files on disk if available. For remote models, remote_url specifies the download location. API-based models use api_endpoint for inference requests. The size_gb and required_memory_gb fields help the system determine if sufficient resources are available before attempting to load a model. The supported_backends list restricts which GPU types can run the model efficiently. Capabilities describe what tasks the model excels at, such as code generation, mathematical reasoning, or general conversation.
The ModelRegistry class manages the collection of available models. It persists the registry to disk as JSON, enabling the configuration to survive application restarts. The load_registry method reconstructs ModelConfig objects from the saved JSON, properly converting string paths back to Path objects and string backend names back to GPUBackend enum values.
Model loading involves several steps including verification that the model exists, checking resource availability, downloading if necessary, and initializing the model with the appropriate backend. Here is the model loader implementation:
import requests
from tqdm import tqdm
import torch
class ModelLoader:
def __init__(self, registry: ModelRegistry, gpu_detector: GPUDetector, cache_dir: Optional[Path] = None):
self.registry = registry
self.gpu_detector = gpu_detector
self.cache_dir = cache_dir or Path.home() / '.llm_app' / 'cache'
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.loaded_models = {}
self.current_backend = gpu_detector.get_optimal_backend()
def download_model(self, model_config: ModelConfig) -> Path:
"""Download model from remote source if not cached"""
if model_config.local_path and model_config.local_path.exists():
return model_config.local_path
if not model_config.remote_url:
raise ValueError(f"No remote URL available for model {model_config.model_id}")
model_cache_path = self.cache_dir / model_config.model_id
model_cache_path.mkdir(parents=True, exist_ok=True)
if model_config.source == 'huggingface':
return self.download_from_huggingface(model_config, model_cache_path)
else:
return self.download_from_url(model_config.remote_url, model_cache_path)
def download_from_huggingface(self, model_config: ModelConfig, cache_path: Path) -> Path:
"""Download model from Hugging Face Hub"""
try:
from huggingface_hub import snapshot_download
snapshot_download(
repo_id=model_config.remote_url,
cache_dir=cache_path,
local_dir=cache_path / 'model',
local_dir_use_symlinks=False
)
final_path = cache_path / 'model'
model_config.local_path = final_path
self.registry.save_registry()
return final_path
except Exception as e:
raise RuntimeError(f"Failed to download from Hugging Face: {str(e)}")
def download_from_url(self, url: str, cache_path: Path) -> Path:
"""Download model from direct URL with progress tracking"""
filename = url.split('/')[-1]
file_path = cache_path / filename
if file_path.exists():
return file_path
response = requests.get(url, stream=True)
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
with open(file_path, 'wb') as f, tqdm(
desc=filename,
total=total_size,
unit='iB',
unit_scale=True,
unit_divisor=1024,
) as progress_bar:
for chunk in response.iter_content(chunk_size=8192):
size = f.write(chunk)
progress_bar.update(size)
return file_path
The ModelLoader class coordinates model acquisition and initialization. The download_model method first checks if the model is already available locally. If not, it determines the source type and delegates to the appropriate download handler. For Hugging Face models, we use the official huggingface_hub library which handles authentication, caching, and incremental downloads efficiently. For direct URL downloads, we implement streaming download with progress tracking using the requests library and tqdm for user feedback.
Loading a model into memory requires backend-specific initialization. Different frameworks and backends have different APIs for model loading. For transformer-based models, we typically use the transformers library from Hugging Face, but the device placement and optimization strategies vary by backend.
def load_model(self, model_id: str, backend: Optional[GPUBackend] = None) -> Any:
"""Load model into memory with appropriate backend"""
if model_id in self.loaded_models:
return self.loaded_models[model_id]
if model_id not in self.registry.models:
raise ValueError(f"Model {model_id} not found in registry")
model_config = self.registry.models[model_id]
target_backend = backend or self.current_backend
if target_backend not in model_config.supported_backends:
raise ValueError(f"Model {model_id} does not support backend {target_backend.value}")
backend_info = self.gpu_detector.backend_info.get(target_backend, {})
if target_backend != GPUBackend.CPU:
available_memory = self.get_available_memory(target_backend, backend_info)
if available_memory < model_config.required_memory_gb:
raise RuntimeError(f"Insufficient memory: need {model_config.required_memory_gb}GB, have {available_memory}GB")
model_path = self.download_model(model_config)
if model_config.model_type == 'transformers':
model = self.load_transformers_model(model_config, model_path, target_backend)
elif model_config.model_type == 'llama_cpp':
model = self.load_llama_cpp_model(model_config, model_path, target_backend)
elif model_config.model_type == 'api':
model = self.create_api_client(model_config)
else:
raise ValueError(f"Unsupported model type: {model_config.model_type}")
self.loaded_models[model_id] = model
return model
def get_available_memory(self, backend: GPUBackend, backend_info: Dict) -> float:
"""Get available GPU memory in GB"""
if backend == GPUBackend.CUDA:
import torch
if torch.cuda.is_available():
device = torch.cuda.current_device()
total = torch.cuda.get_device_properties(device).total_memory
allocated = torch.cuda.memory_allocated(device)
return (total - allocated) / (1024 ** 3)
elif backend == GPUBackend.ROCM:
import torch
if torch.cuda.is_available():
device = torch.cuda.current_device()
total = torch.cuda.get_device_properties(device).total_memory
allocated = torch.cuda.memory_allocated(device)
return (total - allocated) / (1024 ** 3)
elif backend == GPUBackend.MPS:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
return 8.0
return float('inf')
The load_model method orchestrates the entire loading process. It first checks if the model is already loaded to avoid redundant memory usage. Then it validates that the requested backend is supported by the model. Memory availability is checked before proceeding with the actual load operation. The get_available_memory function queries backend-specific APIs to determine free GPU memory. For CUDA and ROCm, PyTorch provides direct memory queries. For MPS, we return a conservative estimate since Apple does not expose detailed memory APIs. The actual model loading is delegated to type-specific loaders.
For transformer models using the Hugging Face transformers library:
def load_transformers_model(self, model_config: ModelConfig, model_path: Path, backend: GPUBackend) -> Any:
"""Load Hugging Face transformers model"""
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
device_map = self.get_device_map(backend)
load_kwargs = {
'pretrained_model_name_or_path': str(model_path),
'device_map': device_map,
'torch_dtype': torch.float16 if backend != GPUBackend.CPU else torch.float32,
}
if model_config.quantization == '8bit':
load_kwargs['load_in_8bit'] = True
elif model_config.quantization == '4bit':
load_kwargs['load_in_4bit'] = True
model = AutoModelForCausalLM.from_pretrained(**load_kwargs)
tokenizer = AutoTokenizer.from_pretrained(str(model_path))
return {'model': model, 'tokenizer': tokenizer, 'type': 'transformers'}
def get_device_map(self, backend: GPUBackend) -> Union[str, Dict]:
"""Get appropriate device map for backend"""
if backend == GPUBackend.CUDA:
return 'auto'
elif backend == GPUBackend.ROCM:
return 'auto'
elif backend == GPUBackend.MPS:
return 'mps'
elif backend == GPUBackend.INTEL:
return 'xpu'
else:
return 'cpu'
The transformers model loader uses AutoModelForCausalLM which automatically selects the appropriate model architecture based on the configuration files in the model directory. The device_map parameter controls where model layers are placed. Setting it to auto enables automatic device placement across multiple GPUs if available. For MPS, we explicitly specify the mps device. The torch_dtype is set to float16 for GPU backends to reduce memory usage and increase speed, while CPU uses float32 for better numerical stability. Quantization options are applied if specified in the model configuration, using the bitsandbytes library integration in transformers.
For llama.cpp based models which are popular for efficient CPU and GPU inference:
def load_llama_cpp_model(self, model_config: ModelConfig, model_path: Path, backend: GPUBackend) -> Any:
"""Load llama.cpp model with appropriate backend"""
try:
from llama_cpp import Llama
except ImportError:
raise ImportError("llama-cpp-python not installed. Install with: pip install llama-cpp-python")
n_gpu_layers = 0
if backend == GPUBackend.CUDA:
n_gpu_layers = -1
elif backend == GPUBackend.ROCM:
n_gpu_layers = -1
elif backend == GPUBackend.MPS:
n_gpu_layers = 1
model_file = self.find_gguf_file(model_path)
llm = Llama(
model_path=str(model_file),
n_gpu_layers=n_gpu_layers,
n_ctx=model_config.context_length,
n_batch=512,
verbose=False
)
return {'model': llm, 'type': 'llama_cpp'}
def find_gguf_file(self, model_path: Path) -> Path:
"""Find GGUF model file in directory"""
if model_path.is_file() and model_path.suffix in ['.gguf', '.bin']:
return model_path
for file in model_path.rglob('*.gguf'):
return file
for file in model_path.rglob('*.bin'):
return file
raise FileNotFoundError(f"No GGUF or BIN model file found in {model_path}")
The llama.cpp loader uses the llama-cpp-python bindings. The n_gpu_layers parameter controls how many transformer layers are offloaded to the GPU. Setting it to negative one offloads all layers. For MPS, we use a conservative value of one layer due to potential stability issues with full offloading on some Apple Silicon configurations. The find_gguf_file helper searches for GGUF format files which are the standard format for llama.cpp models.
For API-based models that run on remote servers:
def create_api_client(self, model_config: ModelConfig) -> Any:
"""Create API client for remote model"""
if not model_config.api_endpoint:
raise ValueError(f"No API endpoint specified for model {model_config.model_id}")
return {
'endpoint': model_config.api_endpoint,
'model_id': model_config.model_id,
'type': 'api'
}
API clients are lightweight wrappers that store the endpoint information. The actual inference happens through HTTP requests, so no model loading is required locally.
AUTOMATIC MODEL SWITCHING AND FAILOVER
A robust LLM application must handle situations where the primary model becomes unavailable due to network issues, insufficient resources, or model file corruption. The failover system maintains a prioritized list of alternative models for each task type and automatically switches to a backup when the primary fails.
from typing import List, Callable
import logging
class ModelFailoverManager:
def __init__(self, model_loader: ModelLoader, registry: ModelRegistry):
self.model_loader = model_loader
self.registry = registry
self.logger = logging.getLogger(__name__)
self.failover_chains = {}
self.initialize_failover_chains()
def initialize_failover_chains(self):
"""Create failover chains for different model categories"""
for model_id, config in self.registry.models.items():
for capability in config.capabilities:
if capability not in self.failover_chains:
self.failover_chains[capability] = []
self.failover_chains[capability].append((config.priority, model_id))
for capability in self.failover_chains:
self.failover_chains[capability].sort(key=lambda x: x[0])
self.failover_chains[capability] = [model_id for _, model_id in self.failover_chains[capability]]
def get_failover_chain(self, primary_model_id: str) -> List[str]:
"""Get ordered list of fallback models for a primary model"""
if primary_model_id not in self.registry.models:
return []
primary_config = self.registry.models[primary_model_id]
candidates = set()
for capability in primary_config.capabilities:
if capability in self.failover_chains:
candidates.update(self.failover_chains[capability])
if primary_model_id in candidates:
candidates.remove(primary_model_id)
ordered_candidates = []
for model_id in candidates:
config = self.registry.models[model_id]
ordered_candidates.append((config.priority, model_id))
ordered_candidates.sort(key=lambda x: x[0])
return [model_id for _, model_id in ordered_candidates]
The ModelFailoverManager maintains failover chains organized by capability. When initializing, it groups models by their capabilities and sorts them by priority. The get_failover_chain method returns an ordered list of alternative models that share capabilities with the primary model. This ensures that if the primary model fails, the system can fall back to a model with similar capabilities.
The actual failover logic wraps model loading and inference with retry mechanisms:
def load_with_failover(self, primary_model_id: str, max_attempts: int = 3) -> Tuple[str, Any]:
"""Attempt to load primary model with automatic failover"""
failover_chain = [primary_model_id] + self.get_failover_chain(primary_model_id)
for attempt, model_id in enumerate(failover_chain):
if attempt >= max_attempts:
break
try:
self.logger.info(f"Attempting to load model: {model_id}")
model = self.model_loader.load_model(model_id)
if attempt > 0:
self.logger.warning(f"Failed over to model: {model_id}")
return model_id, model
except Exception as e:
self.logger.error(f"Failed to load model {model_id}: {str(e)}")
if attempt == len(failover_chain) - 1 or attempt >= max_attempts - 1:
raise RuntimeError(f"All failover attempts exhausted. Last error: {str(e)}")
continue
raise RuntimeError("Failed to load any model in failover chain")
def execute_with_failover(self, primary_model_id: str, inference_func: Callable, *args, **kwargs) -> Any:
"""Execute inference with automatic model failover on errors"""
failover_chain = [primary_model_id] + self.get_failover_chain(primary_model_id)
last_exception = None
for model_id in failover_chain:
try:
model = self.model_loader.loaded_models.get(model_id)
if not model:
model_id, model = self.load_with_failover(model_id, max_attempts=1)
result = inference_func(model, *args, **kwargs)
return result
except Exception as e:
self.logger.error(f"Inference failed with model {model_id}: {str(e)}")
last_exception = e
continue
raise RuntimeError(f"All models in failover chain failed. Last error: {str(last_exception)}")
The load_with_failover method iterates through the failover chain, attempting to load each model in order. If loading succeeds, it returns the model identifier and the loaded model object. If all attempts fail, it raises an exception with details about the last failure. The execute_with_failover method is more sophisticated, wrapping the actual inference function. It attempts inference with each model in the failover chain until one succeeds or all fail. This provides resilience against runtime errors during inference, not just loading failures.
INTELLIGENT MODEL SELECTION BASED ON PROMPTS
The most sophisticated component of the system is the automatic model selector that analyzes user prompts and chooses the most appropriate model. This involves natural language understanding to classify the task type, resource availability checking, and optimization for latency versus quality tradeoffs.
The prompt analyzer extracts features from user input to determine task characteristics:
import re
from typing import Dict, Set
class PromptAnalyzer:
def __init__(self):
self.code_patterns = [
r'```[\w]*\n',
r'def\s+\w+\s*\(',
r'class\s+\w+',
r'function\s+\w+\s*\(',
r'import\s+\w+',
r'#include\s*<',
]
self.math_patterns = [
r'\d+\s*[\+\-\*/]\s*\d+',
r'\\frac\{',
r'\\int_',
r'\\sum_',
r'solve.*equation',
r'calculate.*derivative',
r'prove.*theorem',
]
self.reasoning_keywords = {
'explain', 'why', 'how', 'analyze', 'compare', 'evaluate',
'reasoning', 'logic', 'deduce', 'infer', 'conclude'
}
self.creative_keywords = {
'write', 'story', 'poem', 'creative', 'imagine', 'describe',
'narrative', 'fiction', 'character', 'plot'
}
def analyze_prompt(self, prompt: str) -> Dict[str, any]:
"""Analyze prompt to extract task characteristics"""
features = {
'length': len(prompt),
'has_code': self.detect_code(prompt),
'has_math': self.detect_math(prompt),
'is_reasoning': self.detect_reasoning(prompt),
'is_creative': self.detect_creative(prompt),
'estimated_complexity': self.estimate_complexity(prompt),
'language': self.detect_language(prompt),
}
return features
def detect_code(self, prompt: str) -> bool:
"""Detect if prompt involves code"""
for pattern in self.code_patterns:
if re.search(pattern, prompt, re.IGNORECASE):
return True
code_keywords = ['code', 'program', 'function', 'algorithm', 'debug', 'implement']
prompt_lower = prompt.lower()
return any(keyword in prompt_lower for keyword in code_keywords)
def detect_math(self, prompt: str) -> bool:
"""Detect if prompt involves mathematics"""
for pattern in self.math_patterns:
if re.search(pattern, prompt, re.IGNORECASE):
return True
math_keywords = ['math', 'equation', 'formula', 'calculate', 'solve', 'proof']
prompt_lower = prompt.lower()
return any(keyword in prompt_lower for keyword in math_keywords)
def detect_reasoning(self, prompt: str) -> bool:
"""Detect if prompt requires complex reasoning"""
prompt_lower = prompt.lower()
words = set(prompt_lower.split())
overlap = words.intersection(self.reasoning_keywords)
return len(overlap) >= 2 or any(keyword in prompt_lower for keyword in ['step by step', 'think through'])
def detect_creative(self, prompt: str) -> bool:
"""Detect if prompt is creative writing"""
prompt_lower = prompt.lower()
words = set(prompt_lower.split())
overlap = words.intersection(self.creative_keywords)
return len(overlap) >= 1
def estimate_complexity(self, prompt: str) -> str:
"""Estimate task complexity"""
if len(prompt) > 1000:
return 'high'
complexity_indicators = [
self.detect_code(prompt),
self.detect_math(prompt),
self.detect_reasoning(prompt),
len(prompt) > 500,
]
if sum(complexity_indicators) >= 3:
return 'high'
elif sum(complexity_indicators) >= 1:
return 'medium'
else:
return 'low'
def detect_language(self, prompt: str) -> str:
"""Detect prompt language"""
try:
from langdetect import detect
return detect(prompt)
except:
return 'en'
The PromptAnalyzer uses pattern matching and keyword detection to classify prompts. Code detection looks for syntax patterns like function definitions, import statements, and code blocks. Math detection searches for mathematical notation and keywords. Reasoning and creative writing are identified through keyword sets. Complexity estimation combines multiple signals including prompt length and task type. Language detection uses the langdetect library when available, falling back to English as the default.
The model selector uses these features to score and rank available models:
class IntelligentModelSelector:
def __init__(self, registry: ModelRegistry, model_loader: ModelLoader,
prompt_analyzer: PromptAnalyzer, gpu_detector: GPUDetector):
self.registry = registry
self.model_loader = model_loader
self.prompt_analyzer = prompt_analyzer
self.gpu_detector = gpu_detector
self.logger = logging.getLogger(__name__)
def select_model(self, prompt: str, constraints: Optional[Dict] = None) -> str:
"""Select optimal model for given prompt"""
features = self.prompt_analyzer.analyze_prompt(prompt)
constraints = constraints or {}
max_latency = constraints.get('max_latency_seconds', float('inf'))
min_quality = constraints.get('min_quality_score', 0.0)
preferred_backend = constraints.get('backend', self.gpu_detector.get_optimal_backend())
candidates = self.get_candidate_models(features, preferred_backend)
if not candidates:
raise ValueError("No suitable models found for this prompt")
scored_candidates = []
for model_id in candidates:
score = self.score_model(model_id, features, constraints)
scored_candidates.append((score, model_id))
scored_candidates.sort(reverse=True, key=lambda x: x[0])
selected_model_id = scored_candidates[0][1]
self.logger.info(f"Selected model {selected_model_id} with score {scored_candidates[0][0]:.3f}")
return selected_model_id
def get_candidate_models(self, features: Dict, backend: GPUBackend) -> List[str]:
"""Filter models based on features and backend"""
candidates = []
for model_id, config in self.registry.models.items():
if backend not in config.supported_backends:
continue
if features['has_code'] and 'code' not in config.capabilities:
continue
if features['has_math'] and 'math' not in config.capabilities:
continue
if features['is_reasoning'] and 'reasoning' not in config.capabilities:
continue
candidates.append(model_id)
if not candidates:
candidates = [model_id for model_id, config in self.registry.models.items()
if backend in config.supported_backends]
return candidates
The select_model method orchestrates the selection process. It first analyzes the prompt to extract features, then filters models based on capability requirements and backend compatibility. The get_candidate_models method implements hard constraints, excluding models that lack required capabilities. If no models match all requirements, it falls back to all models supporting the target backend.
Model scoring combines multiple factors:
def score_model(self, model_id: str, features: Dict, constraints: Dict) -> float:
"""Score model suitability for prompt"""
config = self.registry.models[model_id]
score = 0.0
capability_score = self.score_capabilities(config, features)
score += capability_score * 0.4
resource_score = self.score_resources(config, constraints)
score += resource_score * 0.3
latency_score = self.score_latency(config, features, constraints)
score += latency_score * 0.2
priority_score = (100 - config.priority) / 100.0
score += priority_score * 0.1
return score
def score_capabilities(self, config: ModelConfig, features: Dict) -> float:
"""Score model capabilities match"""
score = 0.0
max_score = 0.0
if features['has_code']:
max_score += 1.0
if 'code' in config.capabilities:
score += 1.0
if features['has_math']:
max_score += 1.0
if 'math' in config.capabilities:
score += 1.0
if features['is_reasoning']:
max_score += 1.0
if 'reasoning' in config.capabilities:
score += 1.0
if features['is_creative']:
max_score += 1.0
if 'creative' in config.capabilities:
score += 1.0
if max_score == 0:
return 1.0
return score / max_score
def score_resources(self, config: ModelConfig, constraints: Dict) -> float:
"""Score based on resource availability"""
backend = constraints.get('backend', self.gpu_detector.get_optimal_backend())
if backend == GPUBackend.CPU:
return 1.0
backend_info = self.gpu_detector.backend_info.get(backend, {})
available_memory = self.model_loader.get_available_memory(backend, backend_info)
if available_memory < config.required_memory_gb:
return 0.0
memory_ratio = config.required_memory_gb / available_memory
return 1.0 - (memory_ratio * 0.5)
def score_latency(self, config: ModelConfig, features: Dict, constraints: Dict) -> float:
"""Score based on expected latency"""
max_latency = constraints.get('max_latency_seconds', float('inf'))
estimated_latency = self.estimate_latency(config, features)
if estimated_latency > max_latency:
return 0.0
return 1.0 - (estimated_latency / max_latency)
def estimate_latency(self, config: ModelConfig, features: Dict) -> float:
"""Estimate inference latency in seconds"""
base_latency = config.size_gb * 0.1
if features['estimated_complexity'] == 'high':
base_latency *= 2.0
elif features['estimated_complexity'] == 'medium':
base_latency *= 1.5
if config.quantization in ['8bit', '4bit']:
base_latency *= 0.7
backend = self.gpu_detector.get_optimal_backend()
if backend == GPUBackend.CPU:
base_latency *= 3.0
elif backend == GPUBackend.MPS:
base_latency *= 1.2
return base_latency
The scoring system uses weighted components. Capability matching receives the highest weight at forty percent because using a model specialized for the task type produces better results. Resource availability gets thirty percent weight to ensure the model can actually run. Latency considerations receive twenty percent, and the configured priority gets ten percent. The score_capabilities method computes the overlap between required and available capabilities. The score_resources method checks memory availability and penalizes models that consume a large fraction of available memory. The score_latency method estimates inference time based on model size, complexity, quantization, and backend performance characteristics.
INFERENCE EXECUTION WITH UNIFIED INTERFACE
After selecting the appropriate model, the system needs a unified interface for executing inference regardless of the underlying model type or backend. This abstraction layer handles the differences between transformers models, llama.cpp models, and API-based models.
class UnifiedInferenceEngine:
def __init__(self, model_loader: ModelLoader):
self.model_loader = model_loader
self.logger = logging.getLogger(__name__)
def generate(self, model_id: str, prompt: str, generation_config: Optional[Dict] = None) -> str:
"""Generate response using specified model"""
if model_id not in self.model_loader.loaded_models:
raise ValueError(f"Model {model_id} not loaded")
model_wrapper = self.model_loader.loaded_models[model_id]
model_type = model_wrapper['type']
generation_config = generation_config or {}
if model_type == 'transformers':
return self.generate_transformers(model_wrapper, prompt, generation_config)
elif model_type == 'llama_cpp':
return self.generate_llama_cpp(model_wrapper, prompt, generation_config)
elif model_type == 'api':
return self.generate_api(model_wrapper, prompt, generation_config)
else:
raise ValueError(f"Unknown model type: {model_type}")
def generate_transformers(self, model_wrapper: Dict, prompt: str, config: Dict) -> str:
"""Generate using Hugging Face transformers"""
import torch
model = model_wrapper['model']
tokenizer = model_wrapper['tokenizer']
inputs = tokenizer(prompt, return_tensors='pt')
device = next(model.parameters()).device
inputs = {k: v.to(device) for k, v in inputs.items()}
generation_kwargs = {
'max_new_tokens': config.get('max_tokens', 512),
'temperature': config.get('temperature', 0.7),
'top_p': config.get('top_p', 0.9),
'do_sample': config.get('do_sample', True),
'pad_token_id': tokenizer.pad_token_id or tokenizer.eos_token_id,
}
with torch.no_grad():
outputs = model.generate(**inputs, **generation_kwargs)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
if response.startswith(prompt):
response = response[len(prompt):].strip()
return response
def generate_llama_cpp(self, model_wrapper: Dict, prompt: str, config: Dict) -> str:
"""Generate using llama.cpp"""
model = model_wrapper['model']
generation_kwargs = {
'max_tokens': config.get('max_tokens', 512),
'temperature': config.get('temperature', 0.7),
'top_p': config.get('top_p', 0.9),
'echo': False,
}
response = model(prompt, **generation_kwargs)
return response['choices'][0]['text'].strip()
def generate_api(self, model_wrapper: Dict, prompt: str, config: Dict) -> str:
"""Generate using API endpoint"""
import requests
endpoint = model_wrapper['endpoint']
payload = {
'model': model_wrapper['model_id'],
'prompt': prompt,
'max_tokens': config.get('max_tokens', 512),
'temperature': config.get('temperature', 0.7),
'top_p': config.get('top_p', 0.9),
}
response = requests.post(endpoint, json=payload, timeout=60)
response.raise_for_status()
result = response.json()
if 'choices' in result:
return result['choices'][0]['text'].strip()
elif 'response' in result:
return result['response'].strip()
else:
return str(result)
The UnifiedInferenceEngine provides a consistent generate method regardless of model type. For transformers models, it tokenizes the input, moves tensors to the appropriate device, generates using the model's generate method, and decodes the output. For llama.cpp models, it calls the model directly with the prompt string. For API models, it constructs an HTTP request with the prompt and generation parameters. The generation_config dictionary allows callers to specify parameters like maximum token count, temperature, and top-p sampling.
COMPLETE INTEGRATION AND ORCHESTRATION
The final component ties everything together into a cohesive system that handles the entire workflow from prompt reception to response generation:
class LLMApplication:
def __init__(self, registry_path: Optional[Path] = None, cache_dir: Optional[Path] = None):
self.gpu_detector = GPUDetector()
self.registry = ModelRegistry(registry_path)
self.model_loader = ModelLoader(self.registry, self.gpu_detector, cache_dir)
self.failover_manager = ModelFailoverManager(self.model_loader, self.registry)
self.prompt_analyzer = PromptAnalyzer()
self.model_selector = IntelligentModelSelector(
self.registry, self.model_loader, self.prompt_analyzer, self.gpu_detector
)
self.inference_engine = UnifiedInferenceEngine(self.model_loader)
self.logger = logging.getLogger(__name__)
self.logger.info(f"Initialized with backend: {self.gpu_detector.get_optimal_backend().value}")
def process_prompt(self, prompt: str, model_id: Optional[str] = None,
constraints: Optional[Dict] = None,
generation_config: Optional[Dict] = None) -> Dict:
"""Process prompt with automatic model selection and failover"""
try:
if model_id is None:
model_id = self.model_selector.select_model(prompt, constraints)
self.logger.info(f"Auto-selected model: {model_id}")
if model_id not in self.model_loader.loaded_models:
model_id, _ = self.failover_manager.load_with_failover(model_id)
def inference_func(model_wrapper):
return self.inference_engine.generate(model_id, prompt, generation_config)
response = self.failover_manager.execute_with_failover(
model_id, inference_func
)
return {
'success': True,
'model_id': model_id,
'prompt': prompt,
'response': response,
'backend': self.gpu_detector.get_optimal_backend().value
}
except Exception as e:
self.logger.error(f"Failed to process prompt: {str(e)}")
return {
'success': False,
'error': str(e),
'prompt': prompt
}
def add_model(self, model_config: ModelConfig):
"""Add new model to registry"""
self.registry.models[model_config.model_id] = model_config
self.registry.save_registry()
self.failover_manager.initialize_failover_chains()
def remove_model(self, model_id: str):
"""Remove model from registry and unload if loaded"""
if model_id in self.model_loader.loaded_models:
del self.model_loader.loaded_models[model_id]
if model_id in self.registry.models:
del self.registry.models[model_id]
self.registry.save_registry()
self.failover_manager.initialize_failover_chains()
def list_models(self) -> List[Dict]:
"""List all registered models"""
models = []
for model_id, config in self.registry.models.items():
models.append({
'model_id': model_id,
'type': config.model_type,
'source': config.source,
'size_gb': config.size_gb,
'capabilities': config.capabilities,
'loaded': model_id in self.model_loader.loaded_models
})
return models
def get_system_info(self) -> Dict:
"""Get system information including GPU and loaded models"""
return {
'backend': self.gpu_detector.get_optimal_backend().value,
'available_backends': [b.value for b in self.gpu_detector.available_backends],
'backend_info': {k.value: v for k, v in self.gpu_detector.backend_info.items()},
'loaded_models': list(self.model_loader.loaded_models.keys()),
'registered_models': len(self.registry.models)
}
The LLMApplication class is the main entry point. Its process_prompt method implements the complete workflow. If no model is specified, it uses the intelligent selector. It ensures the model is loaded using the failover manager. It executes inference with automatic failover on errors. The method returns a dictionary containing the response and metadata about which model and backend were used.
The add_model and remove_model methods provide dynamic model management. Adding a model updates the registry and reinitializes failover chains. Removing a model unloads it from memory and updates the registry. The list_models method provides visibility into available models and their status. The get_system_info method returns comprehensive system information useful for debugging and monitoring.
RUNNING EXAMPLE ADDENDUM
The following is a complete, production-ready implementation that integrates all the components discussed above. This code can be deployed directly and supports all GPU backends, local and remote models, automatic failover, and intelligent model selection.
#!/usr/bin/env python3
import sys
import subprocess
import json
import re
import logging
import hashlib
import requests
from pathlib import Path
from typing import Optional, Dict, List, Tuple, Union, Any, Callable, Set
from dataclasses import dataclass, asdict
from enum import Enum
from tqdm import tqdm
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
class GPUBackend(Enum):
"""Enumeration of supported GPU backends"""
CUDA = "cuda"
ROCM = "rocm"
MPS = "mps"
INTEL = "intel"
CPU = "cpu"
class GPUDetector:
"""Detects available GPU hardware and selects optimal backend"""
def __init__(self):
self.available_backends = []
self.backend_info = {}
self.logger = logging.getLogger(self.__class__.__name__)
self.detect_all_backends()
def detect_cuda(self) -> Tuple[bool, Dict]:
"""Detect Nvidia CUDA GPUs and gather information"""
try:
import torch
if torch.cuda.is_available():
device_count = torch.cuda.device_count()
devices = []
for i in range(device_count):
props = torch.cuda.get_device_properties(i)
devices.append({
'index': i,
'name': props.name,
'compute_capability': f"{props.major}.{props.minor}",
'total_memory': props.total_memory,
'multi_processor_count': props.multi_processor_count
})
self.logger.info(f"Detected {device_count} CUDA device(s)")
return True, {'device_count': device_count, 'devices': devices}
except ImportError:
self.logger.debug("PyTorch not available for CUDA detection")
except Exception as e:
self.logger.debug(f"CUDA detection failed: {str(e)}")
return False, {}
def detect_rocm(self) -> Tuple[bool, Dict]:
"""Detect AMD ROCm GPUs"""
try:
import torch
if hasattr(torch.version, 'hip') and torch.version.hip is not None:
if torch.cuda.is_available():
device_count = torch.cuda.device_count()
devices = []
for i in range(device_count):
props = torch.cuda.get_device_properties(i)
devices.append({
'index': i,
'name': props.name,
'total_memory': props.total_memory,
'gcn_arch': props.gcnArchName if hasattr(props, 'gcnArchName') else 'unknown'
})
self.logger.info(f"Detected {device_count} ROCm device(s)")
return True, {'device_count': device_count, 'devices': devices}
except Exception as e:
self.logger.debug(f"ROCm PyTorch detection failed: {str(e)}")
try:
result = subprocess.run(['rocm-smi', '--showproductname'],
capture_output=True, text=True, timeout=5)
if result.returncode == 0 and result.stdout:
self.logger.info("Detected ROCm via rocm-smi")
return True, {'detected_via': 'rocm-smi', 'info': result.stdout.strip()}
except FileNotFoundError:
self.logger.debug("rocm-smi not found")
except Exception as e:
self.logger.debug(f"rocm-smi detection failed: {str(e)}")
return False, {}
def detect_mps(self) -> Tuple[bool, Dict]:
"""Detect Apple Metal Performance Shaders"""
try:
import torch
if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available():
self.logger.info("Detected Apple MPS")
return True, {'backend': 'mps', 'built': torch.backends.mps.is_built()}
except Exception as e:
self.logger.debug(f"MPS detection failed: {str(e)}")
try:
import platform
if platform.system() == 'Darwin' and platform.machine() == 'arm64':
self.logger.info("Detected Apple Silicon platform")
return True, {'platform': 'Apple Silicon', 'detected_via': 'platform'}
except Exception as e:
self.logger.debug(f"Platform detection failed: {str(e)}")
return False, {}
def detect_intel(self) -> Tuple[bool, Dict]:
"""Detect Intel GPUs with oneAPI support"""
try:
import torch
if hasattr(torch, 'xpu') and torch.xpu.is_available():
device_count = torch.xpu.device_count()
devices = []
for i in range(device_count):
devices.append({
'index': i,
'name': torch.xpu.get_device_name(i)
})
self.logger.info(f"Detected {device_count} Intel XPU device(s)")
return True, {'device_count': device_count, 'devices': devices}
except Exception as e:
self.logger.debug(f"Intel XPU detection failed: {str(e)}")
try:
result = subprocess.run(['sycl-ls'], capture_output=True, text=True, timeout=5)
if result.returncode == 0 and 'gpu' in result.stdout.lower():
self.logger.info("Detected Intel GPU via sycl-ls")
return True, {'detected_via': 'sycl-ls', 'info': result.stdout.strip()}
except FileNotFoundError:
self.logger.debug("sycl-ls not found")
except Exception as e:
self.logger.debug(f"sycl-ls detection failed: {str(e)}")
return False, {}
def detect_all_backends(self):
"""Detect all available GPU backends and rank them"""
cuda_available, cuda_info = self.detect_cuda()
if cuda_available:
self.available_backends.append(GPUBackend.CUDA)
self.backend_info[GPUBackend.CUDA] = cuda_info
rocm_available, rocm_info = self.detect_rocm()
if rocm_available and not cuda_available:
self.available_backends.append(GPUBackend.ROCM)
self.backend_info[GPUBackend.ROCM] = rocm_info
mps_available, mps_info = self.detect_mps()
if mps_available:
self.available_backends.append(GPUBackend.MPS)
self.backend_info[GPUBackend.MPS] = mps_info
intel_available, intel_info = self.detect_intel()
if intel_available:
self.available_backends.append(GPUBackend.INTEL)
self.backend_info[GPUBackend.INTEL] = intel_info
self.available_backends.append(GPUBackend.CPU)
self.backend_info[GPUBackend.CPU] = {'fallback': True}
self.logger.info(f"Available backends: {[b.value for b in self.available_backends]}")
def get_optimal_backend(self) -> GPUBackend:
"""Select the optimal backend based on availability and priority"""
priority_order = [GPUBackend.CUDA, GPUBackend.ROCM, GPUBackend.MPS,
GPUBackend.INTEL, GPUBackend.CPU]
for backend in priority_order:
if backend in self.available_backends:
self.logger.info(f"Selected optimal backend: {backend.value}")
return backend
return GPUBackend.CPU
@dataclass
class ModelConfig:
"""Configuration for an LLM model"""
model_id: str
model_type: str
source: str
local_path: Optional[Path]
remote_url: Optional[str]
api_endpoint: Optional[str]
size_gb: float
quantization: Optional[str]
context_length: int
required_memory_gb: float
supported_backends: List[GPUBackend]
capabilities: List[str]
priority: int
class ModelRegistry:
"""Registry for managing available models"""
def __init__(self, registry_path: Optional[Path] = None):
self.registry_path = registry_path or Path.home() / '.llm_app' / 'models.json'
self.models = {}
self.logger = logging.getLogger(self.__class__.__name__)
self.load_registry()
def load_registry(self):
"""Load model registry from disk"""
if self.registry_path.exists():
try:
with open(self.registry_path, 'r') as f:
data = json.load(f)
for model_id, config_dict in data.items():
config_dict['local_path'] = Path(config_dict['local_path']) if config_dict.get('local_path') else None
config_dict['supported_backends'] = [GPUBackend(b) for b in config_dict['supported_backends']]
self.models[model_id] = ModelConfig(**config_dict)
self.logger.info(f"Loaded {len(self.models)} models from registry")
except Exception as e:
self.logger.error(f"Failed to load registry: {str(e)}")
self.initialize_default_registry()
else:
self.initialize_default_registry()
def save_registry(self):
"""Persist model registry to disk"""
self.registry_path.parent.mkdir(parents=True, exist_ok=True)
data = {}
for model_id, config in self.models.items():
config_dict = asdict(config)
config_dict['local_path'] = str(config_dict['local_path']) if config_dict['local_path'] else None
config_dict['supported_backends'] = [b.value for b in config_dict['supported_backends']]
data[model_id] = config_dict
with open(self.registry_path, 'w') as f:
json.dump(data, f, indent=2)
self.logger.info(f"Saved {len(self.models)} models to registry")
def initialize_default_registry(self):
"""Initialize registry with default models"""
self.models = {
'gpt2-small': ModelConfig(
model_id='gpt2-small',
model_type='transformers',
source='huggingface',
local_path=None,
remote_url='gpt2',
api_endpoint=None,
size_gb=0.5,
quantization=None,
context_length=1024,
required_memory_gb=2.0,
supported_backends=[GPUBackend.CUDA, GPUBackend.ROCM, GPUBackend.MPS,
GPUBackend.INTEL, GPUBackend.CPU],
capabilities=['general', 'creative'],
priority=50
),
'tinyllama': ModelConfig(
model_id='tinyllama',
model_type='transformers',
source='huggingface',
local_path=None,
remote_url='TinyLlama/TinyLlama-1.1B-Chat-v1.0',
api_endpoint=None,
size_gb=2.2,
quantization=None,
context_length=2048,
required_memory_gb=4.0,
supported_backends=[GPUBackend.CUDA, GPUBackend.ROCM, GPUBackend.MPS,
GPUBackend.INTEL, GPUBackend.CPU],
capabilities=['general', 'chat', 'reasoning'],
priority=40
),
}
self.save_registry()
self.logger.info("Initialized default model registry")
class ModelLoader:
"""Handles loading models from various sources"""
def __init__(self, registry: ModelRegistry, gpu_detector: GPUDetector, cache_dir: Optional[Path] = None):
self.registry = registry
self.gpu_detector = gpu_detector
self.cache_dir = cache_dir or Path.home() / '.llm_app' / 'cache'
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.loaded_models = {}
self.current_backend = gpu_detector.get_optimal_backend()
self.logger = logging.getLogger(self.__class__.__name__)
def download_model(self, model_config: ModelConfig) -> Path:
"""Download model from remote source if not cached"""
if model_config.local_path and model_config.local_path.exists():
self.logger.info(f"Using local model at {model_config.local_path}")
return model_config.local_path
if not model_config.remote_url:
raise ValueError(f"No remote URL available for model {model_config.model_id}")
model_cache_path = self.cache_dir / model_config.model_id
model_cache_path.mkdir(parents=True, exist_ok=True)
if model_config.source == 'huggingface':
return self.download_from_huggingface(model_config, model_cache_path)
else:
return self.download_from_url(model_config.remote_url, model_cache_path)
def download_from_huggingface(self, model_config: ModelConfig, cache_path: Path) -> Path:
"""Download model from Hugging Face Hub"""
try:
from huggingface_hub import snapshot_download
self.logger.info(f"Downloading {model_config.model_id} from Hugging Face")
snapshot_download(
repo_id=model_config.remote_url,
cache_dir=str(cache_path),
local_dir=str(cache_path / 'model'),
local_dir_use_symlinks=False
)
final_path = cache_path / 'model'
model_config.local_path = final_path
self.registry.save_registry()
self.logger.info(f"Downloaded model to {final_path}")
return final_path
except ImportError:
raise ImportError("huggingface_hub not installed. Install with: pip install huggingface_hub")
except Exception as e:
raise RuntimeError(f"Failed to download from Hugging Face: {str(e)}")
def download_from_url(self, url: str, cache_path: Path) -> Path:
"""Download model from direct URL with progress tracking"""
filename = url.split('/')[-1]
file_path = cache_path / filename
if file_path.exists():
self.logger.info(f"Model already cached at {file_path}")
return file_path
self.logger.info(f"Downloading from {url}")
response = requests.get(url, stream=True)
response.raise_for_status()
total_size = int(response.headers.get('content-length', 0))
with open(file_path, 'wb') as f, tqdm(
desc=filename,
total=total_size,
unit='iB',
unit_scale=True,
unit_divisor=1024,
) as progress_bar:
for chunk in response.iter_content(chunk_size=8192):
size = f.write(chunk)
progress_bar.update(size)
self.logger.info(f"Downloaded to {file_path}")
return file_path
def get_available_memory(self, backend: GPUBackend, backend_info: Dict) -> float:
"""Get available GPU memory in GB"""
if backend == GPUBackend.CUDA:
try:
import torch
if torch.cuda.is_available():
device = torch.cuda.current_device()
total = torch.cuda.get_device_properties(device).total_memory
allocated = torch.cuda.memory_allocated(device)
available_bytes = total - allocated
return available_bytes / (1024 ** 3)
except Exception as e:
self.logger.warning(f"Failed to get CUDA memory: {str(e)}")
elif backend == GPUBackend.ROCM:
try:
import torch
if torch.cuda.is_available():
device = torch.cuda.current_device()
total = torch.cuda.get_device_properties(device).total_memory
allocated = torch.cuda.memory_allocated(device)
available_bytes = total - allocated
return available_bytes / (1024 ** 3)
except Exception as e:
self.logger.warning(f"Failed to get ROCm memory: {str(e)}")
elif backend == GPUBackend.MPS:
return 8.0
return float('inf')
def load_model(self, model_id: str, backend: Optional[GPUBackend] = None) -> Any:
"""Load model into memory with appropriate backend"""
if model_id in self.loaded_models:
self.logger.info(f"Model {model_id} already loaded")
return self.loaded_models[model_id]
if model_id not in self.registry.models:
raise ValueError(f"Model {model_id} not found in registry")
model_config = self.registry.models[model_id]
target_backend = backend or self.current_backend
if target_backend not in model_config.supported_backends:
raise ValueError(f"Model {model_id} does not support backend {target_backend.value}")
backend_info = self.gpu_detector.backend_info.get(target_backend, {})
if target_backend != GPUBackend.CPU:
available_memory = self.get_available_memory(target_backend, backend_info)
if available_memory < model_config.required_memory_gb:
raise RuntimeError(f"Insufficient memory: need {model_config.required_memory_gb}GB, have {available_memory:.1f}GB")
self.logger.info(f"Loading model {model_id} on {target_backend.value}")
model_path = self.download_model(model_config)
if model_config.model_type == 'transformers':
model = self.load_transformers_model(model_config, model_path, target_backend)
elif model_config.model_type == 'llama_cpp':
model = self.load_llama_cpp_model(model_config, model_path, target_backend)
elif model_config.model_type == 'api':
model = self.create_api_client(model_config)
else:
raise ValueError(f"Unsupported model type: {model_config.model_type}")
self.loaded_models[model_id] = model
self.logger.info(f"Successfully loaded model {model_id}")
return model
def get_device_map(self, backend: GPUBackend) -> Union[str, Dict]:
"""Get appropriate device map for backend"""
if backend == GPUBackend.CUDA:
return 'auto'
elif backend == GPUBackend.ROCM:
return 'auto'
elif backend == GPUBackend.MPS:
return 'mps'
elif backend == GPUBackend.INTEL:
return 'xpu'
else:
return 'cpu'
def load_transformers_model(self, model_config: ModelConfig, model_path: Path, backend: GPUBackend) -> Any:
"""Load Hugging Face transformers model"""
try:
from transformers import AutoModelForCausalLM, AutoTokenizer
import torch
except ImportError:
raise ImportError("transformers and torch not installed. Install with: pip install transformers torch")
device_map = self.get_device_map(backend)
load_kwargs = {
'pretrained_model_name_or_path': str(model_path),
'device_map': device_map,
'torch_dtype': torch.float16 if backend != GPUBackend.CPU else torch.float32,
'low_cpu_mem_usage': True,
}
if model_config.quantization == '8bit':
load_kwargs['load_in_8bit'] = True
elif model_config.quantization == '4bit':
load_kwargs['load_in_4bit'] = True
model = AutoModelForCausalLM.from_pretrained(**load_kwargs)
tokenizer = AutoTokenizer.from_pretrained(str(model_path))
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
return {'model': model, 'tokenizer': tokenizer, 'type': 'transformers'}
def find_gguf_file(self, model_path: Path) -> Path:
"""Find GGUF model file in directory"""
if model_path.is_file() and model_path.suffix in ['.gguf', '.bin']:
return model_path
for file in model_path.rglob('*.gguf'):
return file
for file in model_path.rglob('*.bin'):
return file
raise FileNotFoundError(f"No GGUF or BIN model file found in {model_path}")
def load_llama_cpp_model(self, model_config: ModelConfig, model_path: Path, backend: GPUBackend) -> Any:
"""Load llama.cpp model with appropriate backend"""
try:
from llama_cpp import Llama
except ImportError:
raise ImportError("llama-cpp-python not installed. Install with: pip install llama-cpp-python")
n_gpu_layers = 0
if backend == GPUBackend.CUDA:
n_gpu_layers = -1
elif backend == GPUBackend.ROCM:
n_gpu_layers = -1
elif backend == GPUBackend.MPS:
n_gpu_layers = 1
model_file = self.find_gguf_file(model_path)
llm = Llama(
model_path=str(model_file),
n_gpu_layers=n_gpu_layers,
n_ctx=model_config.context_length,
n_batch=512,
verbose=False
)
return {'model': llm, 'type': 'llama_cpp'}
def create_api_client(self, model_config: ModelConfig) -> Any:
"""Create API client for remote model"""
if not model_config.api_endpoint:
raise ValueError(f"No API endpoint specified for model {model_config.model_id}")
return {
'endpoint': model_config.api_endpoint,
'model_id': model_config.model_id,
'type': 'api'
}
def unload_model(self, model_id: str):
"""Unload model from memory"""
if model_id in self.loaded_models:
del self.loaded_models[model_id]
self.logger.info(f"Unloaded model {model_id}")
try:
import torch
if torch.cuda.is_available():
torch.cuda.empty_cache()
except:
pass
class ModelFailoverManager:
"""Manages automatic model failover"""
def __init__(self, model_loader: ModelLoader, registry: ModelRegistry):
self.model_loader = model_loader
self.registry = registry
self.logger = logging.getLogger(self.__class__.__name__)
self.failover_chains = {}
self.initialize_failover_chains()
def initialize_failover_chains(self):
"""Create failover chains for different model categories"""
for model_id, config in self.registry.models.items():
for capability in config.capabilities:
if capability not in self.failover_chains:
self.failover_chains[capability] = []
self.failover_chains[capability].append((config.priority, model_id))
for capability in self.failover_chains:
self.failover_chains[capability].sort(key=lambda x: x[0])
self.failover_chains[capability] = [model_id for _, model_id in self.failover_chains[capability]]
self.logger.info(f"Initialized failover chains for {len(self.failover_chains)} capabilities")
def get_failover_chain(self, primary_model_id: str) -> List[str]:
"""Get ordered list of fallback models for a primary model"""
if primary_model_id not in self.registry.models:
return []
primary_config = self.registry.models[primary_model_id]
candidates = set()
for capability in primary_config.capabilities:
if capability in self.failover_chains:
candidates.update(self.failover_chains[capability])
if primary_model_id in candidates:
candidates.remove(primary_model_id)
ordered_candidates = []
for model_id in candidates:
config = self.registry.models[model_id]
ordered_candidates.append((config.priority, model_id))
ordered_candidates.sort(key=lambda x: x[0])
return [model_id for _, model_id in ordered_candidates]
def load_with_failover(self, primary_model_id: str, max_attempts: int = 3) -> Tuple[str, Any]:
"""Attempt to load primary model with automatic failover"""
failover_chain = [primary_model_id] + self.get_failover_chain(primary_model_id)
for attempt, model_id in enumerate(failover_chain):
if attempt >= max_attempts:
break
try:
self.logger.info(f"Attempting to load model: {model_id}")
model = self.model_loader.load_model(model_id)
if attempt > 0:
self.logger.warning(f"Failed over to model: {model_id}")
return model_id, model
except Exception as e:
self.logger.error(f"Failed to load model {model_id}: {str(e)}")
if attempt == len(failover_chain) - 1 or attempt >= max_attempts - 1:
raise RuntimeError(f"All failover attempts exhausted. Last error: {str(e)}")
continue
raise RuntimeError("Failed to load any model in failover chain")
def execute_with_failover(self, primary_model_id: str, inference_func: Callable, *args, **kwargs) -> Any:
"""Execute inference with automatic model failover on errors"""
failover_chain = [primary_model_id] + self.get_failover_chain(primary_model_id)
last_exception = None
for model_id in failover_chain:
try:
model = self.model_loader.loaded_models.get(model_id)
if not model:
model_id, model = self.load_with_failover(model_id, max_attempts=1)
result = inference_func(model, *args, **kwargs)
return result
except Exception as e:
self.logger.error(f"Inference failed with model {model_id}: {str(e)}")
last_exception = e
continue
raise RuntimeError(f"All models in failover chain failed. Last error: {str(last_exception)}")
class PromptAnalyzer:
"""Analyzes prompts to determine task characteristics"""
def __init__(self):
self.code_patterns = [
r'```[\w]*\n',
r'def\s+\w+\s*\(',
r'class\s+\w+',
r'function\s+\w+\s*\(',
r'import\s+\w+',
r'#include\s*<',
r'public\s+class',
r'fn\s+\w+\s*\(',
]
self.math_patterns = [
r'\d+\s*[\+\-\*/]\s*\d+',
r'\\frac\{',
r'\\int_',
r'\\sum_',
r'solve.*equation',
r'calculate.*derivative',
r'prove.*theorem',
r'\d+\^\d+',
]
self.reasoning_keywords = {
'explain', 'why', 'how', 'analyze', 'compare', 'evaluate',
'reasoning', 'logic', 'deduce', 'infer', 'conclude', 'because',
'therefore', 'thus', 'hence', 'consequently'
}
self.creative_keywords = {
'write', 'story', 'poem', 'creative', 'imagine', 'describe',
'narrative', 'fiction', 'character', 'plot', 'scene', 'dialogue'
}
self.logger = logging.getLogger(self.__class__.__name__)
def analyze_prompt(self, prompt: str) -> Dict[str, any]:
"""Analyze prompt to extract task characteristics"""
features = {
'length': len(prompt),
'has_code': self.detect_code(prompt),
'has_math': self.detect_math(prompt),
'is_reasoning': self.detect_reasoning(prompt),
'is_creative': self.detect_creative(prompt),
'estimated_complexity': self.estimate_complexity(prompt),
'language': self.detect_language(prompt),
}
self.logger.debug(f"Prompt features: {features}")
return features
def detect_code(self, prompt: str) -> bool:
"""Detect if prompt involves code"""
for pattern in self.code_patterns:
if re.search(pattern, prompt, re.IGNORECASE):
return True
code_keywords = ['code', 'program', 'function', 'algorithm', 'debug', 'implement',
'script', 'syntax', 'compile', 'execute']
prompt_lower = prompt.lower()
return any(keyword in prompt_lower for keyword in code_keywords)
def detect_math(self, prompt: str) -> bool:
"""Detect if prompt involves mathematics"""
for pattern in self.math_patterns:
if re.search(pattern, prompt, re.IGNORECASE):
return True
math_keywords = ['math', 'equation', 'formula', 'calculate', 'solve', 'proof',
'algebra', 'calculus', 'geometry', 'statistics', 'probability']
prompt_lower = prompt.lower()
return any(keyword in prompt_lower for keyword in math_keywords)
def detect_reasoning(self, prompt: str) -> bool:
"""Detect if prompt requires complex reasoning"""
prompt_lower = prompt.lower()
words = set(prompt_lower.split())
overlap = words.intersection(self.reasoning_keywords)
return len(overlap) >= 2 or any(phrase in prompt_lower for phrase in ['step by step', 'think through', 'let\'s think'])
def detect_creative(self, prompt: str) -> bool:
"""Detect if prompt is creative writing"""
prompt_lower = prompt.lower()
words = set(prompt_lower.split())
overlap = words.intersection(self.creative_keywords)
return len(overlap) >= 1
def estimate_complexity(self, prompt: str) -> str:
"""Estimate task complexity"""
if len(prompt) > 1000:
return 'high'
complexity_indicators = [
self.detect_code(prompt),
self.detect_math(prompt),
self.detect_reasoning(prompt),
len(prompt) > 500,
]
indicator_count = sum(complexity_indicators)
if indicator_count >= 3:
return 'high'
elif indicator_count >= 1:
return 'medium'
else:
return 'low'
def detect_language(self, prompt: str) -> str:
"""Detect prompt language"""
try:
from langdetect import detect
return detect(prompt)
except:
return 'en'
class IntelligentModelSelector:
"""Selects optimal model based on prompt analysis"""
def __init__(self, registry: ModelRegistry, model_loader: ModelLoader,
prompt_analyzer: PromptAnalyzer, gpu_detector: GPUDetector):
self.registry = registry
self.model_loader = model_loader
self.prompt_analyzer = prompt_analyzer
self.gpu_detector = gpu_detector
self.logger = logging.getLogger(self.__class__.__name__)
def select_model(self, prompt: str, constraints: Optional[Dict] = None) -> str:
"""Select optimal model for given prompt"""
features = self.prompt_analyzer.analyze_prompt(prompt)
constraints = constraints or {}
max_latency = constraints.get('max_latency_seconds', float('inf'))
min_quality = constraints.get('min_quality_score', 0.0)
preferred_backend = constraints.get('backend', self.gpu_detector.get_optimal_backend())
candidates = self.get_candidate_models(features, preferred_backend)
if not candidates:
self.logger.warning("No suitable models found, using all available models")
candidates = list(self.registry.models.keys())
if not candidates:
raise ValueError("No models available in registry")
scored_candidates = []
for model_id in candidates:
score = self.score_model(model_id, features, constraints)
scored_candidates.append((score, model_id))
scored_candidates.sort(reverse=True, key=lambda x: x[0])
selected_model_id = scored_candidates[0][1]
self.logger.info(f"Selected model {selected_model_id} with score {scored_candidates[0][0]:.3f}")
return selected_model_id
def get_candidate_models(self, features: Dict, backend: GPUBackend) -> List[str]:
"""Filter models based on features and backend"""
candidates = []
for model_id, config in self.registry.models.items():
if backend not in config.supported_backends:
continue
if features['has_code'] and 'code' not in config.capabilities:
continue
if features['has_math'] and 'math' not in config.capabilities:
continue
if features['is_reasoning'] and 'reasoning' not in config.capabilities:
continue
candidates.append(model_id)
if not candidates:
candidates = [model_id for model_id, config in self.registry.models.items()
if backend in config.supported_backends]
return candidates
def score_model(self, model_id: str, features: Dict, constraints: Dict) -> float:
"""Score model suitability for prompt"""
config = self.registry.models[model_id]
score = 0.0
capability_score = self.score_capabilities(config, features)
score += capability_score * 0.4
resource_score = self.score_resources(config, constraints)
score += resource_score * 0.3
latency_score = self.score_latency(config, features, constraints)
score += latency_score * 0.2
priority_score = (100 - config.priority) / 100.0
score += priority_score * 0.1
return score
def score_capabilities(self, config: ModelConfig, features: Dict) -> float:
"""Score model capabilities match"""
score = 0.0
max_score = 0.0
if features['has_code']:
max_score += 1.0
if 'code' in config.capabilities:
score += 1.0
if features['has_math']:
max_score += 1.0
if 'math' in config.capabilities:
score += 1.0
if features['is_reasoning']:
max_score += 1.0
if 'reasoning' in config.capabilities:
score += 1.0
if features['is_creative']:
max_score += 1.0
if 'creative' in config.capabilities:
score += 1.0
if max_score == 0:
return 1.0
return score / max_score
def score_resources(self, config: ModelConfig, constraints: Dict) -> float:
"""Score based on resource availability"""
backend = constraints.get('backend', self.gpu_detector.get_optimal_backend())
if backend == GPUBackend.CPU:
return 1.0
backend_info = self.gpu_detector.backend_info.get(backend, {})
available_memory = self.model_loader.get_available_memory(backend, backend_info)
if available_memory < config.required_memory_gb:
return 0.0
memory_ratio = config.required_memory_gb / available_memory
return 1.0 - (memory_ratio * 0.5)
def score_latency(self, config: ModelConfig, features: Dict, constraints: Dict) -> float:
"""Score based on expected latency"""
max_latency = constraints.get('max_latency_seconds', float('inf'))
estimated_latency = self.estimate_latency(config, features)
if estimated_latency > max_latency:
return 0.0
return 1.0 - (estimated_latency / max_latency)
def estimate_latency(self, config: ModelConfig, features: Dict) -> float:
"""Estimate inference latency in seconds"""
base_latency = config.size_gb * 0.1
if features['estimated_complexity'] == 'high':
base_latency *= 2.0
elif features['estimated_complexity'] == 'medium':
base_latency *= 1.5
if config.quantization in ['8bit', '4bit']:
base_latency *= 0.7
backend = self.gpu_detector.get_optimal_backend()
if backend == GPUBackend.CPU:
base_latency *= 3.0
elif backend == GPUBackend.MPS:
base_latency *= 1.2
return base_latency
class UnifiedInferenceEngine:
"""Unified interface for model inference"""
def __init__(self, model_loader: ModelLoader):
self.model_loader = model_loader
self.logger = logging.getLogger(self.__class__.__name__)
def generate(self, model_id: str, prompt: str, generation_config: Optional[Dict] = None) -> str:
"""Generate response using specified model"""
if model_id not in self.model_loader.loaded_models:
raise ValueError(f"Model {model_id} not loaded")
model_wrapper = self.model_loader.loaded_models[model_id]
model_type = model_wrapper['type']
generation_config = generation_config or {}
self.logger.info(f"Generating with model {model_id} (type: {model_type})")
if model_type == 'transformers':
return self.generate_transformers(model_wrapper, prompt, generation_config)
elif model_type == 'llama_cpp':
return self.generate_llama_cpp(model_wrapper, prompt, generation_config)
elif model_type == 'api':
return self.generate_api(model_wrapper, prompt, generation_config)
else:
raise ValueError(f"Unknown model type: {model_type}")
def generate_transformers(self, model_wrapper: Dict, prompt: str, config: Dict) -> str:
"""Generate using Hugging Face transformers"""
import torch
model = model_wrapper['model']
tokenizer = model_wrapper['tokenizer']
inputs = tokenizer(prompt, return_tensors='pt', padding=True, truncation=True)
device = next(model.parameters()).device
inputs = {k: v.to(device) for k, v in inputs.items()}
generation_kwargs = {
'max_new_tokens': config.get('max_tokens', 512),
'temperature': config.get('temperature', 0.7),
'top_p': config.get('top_p', 0.9),
'do_sample': config.get('do_sample', True),
'pad_token_id': tokenizer.pad_token_id or tokenizer.eos_token_id,
}
with torch.no_grad():
outputs = model.generate(**inputs, **generation_kwargs)
response = tokenizer.decode(outputs[0], skip_special_tokens=True)
if response.startswith(prompt):
response = response[len(prompt):].strip()
return response
def generate_llama_cpp(self, model_wrapper: Dict, prompt: str, config: Dict) -> str:
"""Generate using llama.cpp"""
model = model_wrapper['model']
generation_kwargs = {
'max_tokens': config.get('max_tokens', 512),
'temperature': config.get('temperature', 0.7),
'top_p': config.get('top_p', 0.9),
'echo': False,
}
response = model(prompt, **generation_kwargs)
return response['choices'][0]['text'].strip()
def generate_api(self, model_wrapper: Dict, prompt: str, config: Dict) -> str:
"""Generate using API endpoint"""
import requests
endpoint = model_wrapper['endpoint']
payload = {
'model': model_wrapper['model_id'],
'prompt': prompt,
'max_tokens': config.get('max_tokens', 512),
'temperature': config.get('temperature', 0.7),
'top_p': config.get('top_p', 0.9),
}
response = requests.post(endpoint, json=payload, timeout=60)
response.raise_for_status()
result = response.json()
if 'choices' in result:
return result['choices'][0]['text'].strip()
elif 'response' in result:
return result['response'].strip()
else:
return str(result)
class LLMApplication:
"""Main application class integrating all components"""
def __init__(self, registry_path: Optional[Path] = None, cache_dir: Optional[Path] = None):
self.logger = logging.getLogger(self.__class__.__name__)
self.logger.info("Initializing LLM Application")
self.gpu_detector = GPUDetector()
self.registry = ModelRegistry(registry_path)
self.model_loader = ModelLoader(self.registry, self.gpu_detector, cache_dir)
self.failover_manager = ModelFailoverManager(self.model_loader, self.registry)
self.prompt_analyzer = PromptAnalyzer()
self.model_selector = IntelligentModelSelector(
self.registry, self.model_loader, self.prompt_analyzer, self.gpu_detector
)
self.inference_engine = UnifiedInferenceEngine(self.model_loader)
self.logger.info(f"Initialized with backend: {self.gpu_detector.get_optimal_backend().value}")
def process_prompt(self, prompt: str, model_id: Optional[str] = None,
constraints: Optional[Dict] = None,
generation_config: Optional[Dict] = None) -> Dict:
"""Process prompt with automatic model selection and failover"""
try:
if model_id is None:
model_id = self.model_selector.select_model(prompt, constraints)
self.logger.info(f"Auto-selected model: {model_id}")
if model_id not in self.model_loader.loaded_models:
model_id, _ = self.failover_manager.load_with_failover(model_id)
def inference_func(model_wrapper):
return self.inference_engine.generate(model_id, prompt, generation_config)
response = self.failover_manager.execute_with_failover(
model_id, inference_func
)
return {
'success': True,
'model_id': model_id,
'prompt': prompt,
'response': response,
'backend': self.gpu_detector.get_optimal_backend().value
}
except Exception as e:
self.logger.error(f"Failed to process prompt: {str(e)}")
return {
'success': False,
'error': str(e),
'prompt': prompt
}
def add_model(self, model_config: ModelConfig):
"""Add new model to registry"""
self.registry.models[model_config.model_id] = model_config
self.registry.save_registry()
self.failover_manager.initialize_failover_chains()
self.logger.info(f"Added model {model_config.model_id} to registry")
def remove_model(self, model_id: str):
"""Remove model from registry and unload if loaded"""
if model_id in self.model_loader.loaded_models:
self.model_loader.unload_model(model_id)
if model_id in self.registry.models:
del self.registry.models[model_id]
self.registry.save_registry()
self.failover_manager.initialize_failover_chains()
self.logger.info(f"Removed model {model_id} from registry")
def list_models(self) -> List[Dict]:
"""List all registered models"""
models = []
for model_id, config in self.registry.models.items():
models.append({
'model_id': model_id,
'type': config.model_type,
'source': config.source,
'size_gb': config.size_gb,
'capabilities': config.capabilities,
'loaded': model_id in self.model_loader.loaded_models,
'priority': config.priority
})
return models
def get_system_info(self) -> Dict:
"""Get system information including GPU and loaded models"""
return {
'backend': self.gpu_detector.get_optimal_backend().value,
'available_backends': [b.value for b in self.gpu_detector.available_backends],
'backend_info': {k.value: v for k, v in self.gpu_detector.backend_info.items()},
'loaded_models': list(self.model_loader.loaded_models.keys()),
'registered_models': len(self.registry.models)
}
def main():
"""Example usage of the LLM Application"""
app = LLMApplication()
print("System Information:")
print(json.dumps(app.get_system_info(), indent=2))
print("\n" + "="*80 + "\n")
print("Available Models:")
for model in app.list_models():
print(f" - {model['model_id']}: {model['capabilities']} (loaded: {model['loaded']})")
print("\n" + "="*80 + "\n")
test_prompts = [
"Explain how photosynthesis works in plants.",
"Write a Python function to calculate fibonacci numbers.",
"Solve the equation: 2x + 5 = 15",
]
for i, prompt in enumerate(test_prompts, 1):
print(f"Test {i}: {prompt}")
result = app.process_prompt(prompt)
if result['success']:
print(f"Model: {result['model_id']}")
print(f"Backend: {result['backend']}")
print(f"Response: {result['response'][:200]}...")
else:
print(f"Error: {result['error']}")
print("\n" + "="*80 + "\n")
if __name__ == "__main__":
main()
This complete implementation provides a production-ready system for intelligent LLM application deployment. The code handles GPU detection across all major vendors, supports multiple model formats and sources, implements robust failover mechanisms, and automatically selects optimal models based on prompt analysis. The system is extensible, allowing new models to be added dynamically, and provides comprehensive logging and error handling throughout. All components follow clean architecture principles with clear separation of concerns and well-defined interfaces between modules.
