AI Architect Expert

Expert guidance for designing AI systems, MLOps architecture, scalable ML infrastructure, and AI platform engineering.

Core Concepts AI System Architecture Model serving architectures Real-time vs batch inference Feature stores Model registries Training pipelines Data versioning MLOps Infrastructure CI/CD for ML Model monitoring and observability A/B testing frameworks Model retraining automation Resource orchestration Cost optimization Scalability Patterns Distributed training Model parallelism Data parallelism Inference optimization Caching strategies Load balancing ML Platform Architecture from dataclasses import dataclass from typing import Dict, List, Optional from enum import Enum

class ModelStage(Enum): DEVELOPMENT = "development" STAGING = "staging" PRODUCTION = "production" ARCHIVED = "archived"

@dataclass class ModelMetadata: name: str version: str framework: str stage: ModelStage metrics: Dict[str, float] created_at: str updated_at: str

class ModelRegistry: """Central model registry for ML platform"""

def __init__(self):
    self.models: Dict[str, List[ModelMetadata]] = {}

def register_model(self, model: ModelMetadata) -> str:
    """Register new model version"""
    if model.name not in self.models:
        self.models[model.name] = []

    self.models[model.name].append(model)
    return f"{model.name}:{model.version}"

def promote_model(self, name: str, version: str, stage: ModelStage):
    """Promote model to different stage"""
    for model in self.models.get(name, []):
        if model.version == version:
            model.stage = stage
            return True
    return False

def get_production_model(self, name: str) -> Optional[ModelMetadata]:
    """Get current production model"""
    for model in self.models.get(name, []):
        if model.stage == ModelStage.PRODUCTION:
            return model
    return None

class FeatureStore: """Feature store for ML features"""

def __init__(self):
    self.features: Dict[str, Dict] = {}
    self.feature_groups: Dict[str, List[str]] = {}

def register_feature(self, name: str, dtype: str, description: str,
                    transformation: Optional[str] = None):
    """Register feature definition"""
    self.features[name] = {
        "dtype": dtype,
        "description": description,
        "transformation": transformation
    }

def create_feature_group(self, group_name: str, feature_names: List[str]):
    """Create feature group for reuse"""
    self.feature_groups[group_name] = feature_names

def get_features(self, entity_id: str, feature_names: List[str]) -> Dict:
    """Retrieve feature values for entity"""
    # In production, this would query online/offline stores
    return {name: self._fetch_feature(entity_id, name)
            for name in feature_names}

Training Pipeline Architecture from abc import ABC, abstractmethod import torch.distributed as dist

class TrainingPipeline(ABC): """Base training pipeline"""

def __init__(self, config: Dict):
    self.config = config
    self.experiment_tracker = None
    self.checkpointer = None

@abstractmethod
def prepare_data(self):
    """Data preparation step"""
    pass

@abstractmethod
def train(self):
    """Training step"""
    pass

@abstractmethod
def evaluate(self):
    """Evaluation step"""
    pass

def run(self):
    """Execute full pipeline"""
    self.prepare_data()
    self.train()
    metrics = self.evaluate()
    self.log_metrics(metrics)
    return metrics

class DistributedTrainingPipeline(TrainingPipeline): """Distributed training with DDP"""

def __init__(self, config: Dict, world_size: int, rank: int):
    super().__init__(config)
    self.world_size = world_size
    self.rank = rank
    self.setup_distributed()

def setup_distributed(self):
    """Initialize distributed training"""
    dist.init_process_group(
        backend='nccl',
        world_size=self.world_size,
        rank=self.rank
    )

def prepare_data(self):
    """Distribute data across workers"""
    from torch.utils.data.distributed import DistributedSampler

    self.sampler = DistributedSampler(
        self.dataset,
        num_replicas=self.world_size,
        rank=self.rank
    )

def train(self):
    """Distributed training loop"""
    from torch.nn.parallel import DistributedDataParallel as DDP

    model = DDP(self.model, device_ids=[self.rank])

    for epoch in range(self.config['epochs']):
        self.sampler.set_epoch(epoch)

        for batch in self.dataloader:
            loss = self.train_step(model, batch)

            if self.rank == 0:
                self.log_loss(loss)

Model Serving Architecture from fastapi import FastAPI, BackgroundTasks from prometheus_client import Counter, Histogram import asyncio

Metrics

prediction_counter = Counter('predictions_total', 'Total predictions') prediction_latency = Histogram('prediction_latency_seconds', 'Prediction latency')

class ModelServer: """Production model serving"""

def __init__(self, model_registry: ModelRegistry):
    self.registry = model_registry
    self.loaded_models = {}
    self.prediction_cache = {}

async def load_model(self, name: str, version: str = "production"):
    """Load model into memory"""
    if version == "production":
        model_metadata = self.registry.get_production_model(name)
    else:
        model_metadata = self.registry.get_model(name, version)

    if not model_metadata:
        raise ValueError(f"Model {name}:{version} not found")

    # Load model from storage
    model = await self._load_from_storage(model_metadata)
    self.loaded_models[f"{name}:{version}"] = model

    return model

@prediction_latency.time()
async def predict(self, model_name: str, features: Dict) -> Dict:
    """Make prediction with caching"""
    prediction_counter.inc()

    # Check cache
    cache_key = self._generate_cache_key(model_name, features)
    if cache_key in self.prediction_cache:
        return self.prediction_cache[cache_key]

    # Get model
    model = self.loaded_models.get(model_name)
    if not model:
        model = await self.load_model(model_name)

    # Predict
    result = await self._run_inference(model, features)

    # Cache result
    self.prediction_cache[cache_key] = result

    return result

async def predict_batch(self, model_name: str,
                       batch_features: List[Dict]) -> List[Dict]:
    """Batch prediction for efficiency"""
    tasks = [self.predict(model_name, features)
            for features in batch_features]
    return await asyncio.gather(*tasks)

app = FastAPI() model_server = ModelServer(model_registry=ModelRegistry())

@app.post("/predict/{model_name}") async def predict_endpoint(model_name: str, features: Dict): return await model_server.predict(model_name, features)

Monitoring and Observability from dataclasses import dataclass from datetime import datetime import numpy as np

@dataclass class PredictionLog: timestamp: datetime model_name: str model_version: str features: Dict prediction: any latency_ms: float input_hash: str

class ModelMonitor: """Monitor model performance in production"""

def __init__(self):
    self.logs: List[PredictionLog] = []
    self.metrics = {}

def log_prediction(self, log: PredictionLog):
    """Log prediction for monitoring"""
    self.logs.append(log)

    # Update metrics
    self.update_latency_metrics(log)
    self.check_data_drift(log)

def update_latency_metrics(self, log: PredictionLog):
    """Track prediction latency"""
    model_key = f"{log.model_name}:{log.model_version}"

    if model_key not in self.metrics:
        self.metrics[model_key] = {
            "latencies": [],
            "predictions": 0
        }

    self.metrics[model_key]["latencies"].append(log.latency_ms)
    self.metrics[model_key]["predictions"] += 1

def check_data_drift(self, log: PredictionLog):
    """Detect data drift in input features"""
    # Compare current feature distributions with training data
    # Alert if significant drift detected
    pass

def get_model_health(self, model_name: str) -> Dict:
    """Get model health metrics"""
    model_metrics = self.metrics.get(model_name, {})

    latencies = model_metrics.get("latencies", [])

    return {
        "total_predictions": model_metrics.get("predictions", 0),
        "avg_latency_ms": np.mean(latencies) if latencies else 0,
        "p95_latency_ms": np.percentile(latencies, 95) if latencies else 0,
        "p99_latency_ms": np.percentile(latencies, 99) if latencies else 0
    }

Best Practices Architecture Design Separate training and serving infrastructure Use feature stores for consistency Implement model versioning from day one Design for horizontal scalability Plan for model rollback capability Build monitoring into the architecture MLOps Automate model retraining pipelines Implement CI/CD for models Version everything (data, code, models) Monitor model performance metrics Track data drift and model decay Implement gradual rollout (canary/blue-green) Infrastructure Use GPU efficiently (batching, mixed precision) Implement caching for repeated predictions Consider model compression (quantization, pruning) Plan for disaster recovery Optimize costs (spot instances, autoscaling) Use managed services where appropriate Anti-Patterns

❌ No model versioning or registry ❌ Training and serving environment mismatch ❌ No monitoring or alerting ❌ Manual model deployment process ❌ Ignoring data drift ❌ No rollback strategy ❌ Over-engineering for initial MVP

Resources MLflow: https://mlflow.org/ Kubeflow: https://www.kubeflow.org/ Seldon Core: https://www.seldon.io/ BentoML: https://www.bentoml.com/ Weights & Biases: https://wandb.ai/