ai-architect-expert

name	ai-architect-expert
version	1.0.0
description	Expert-level AI system design, MLOps, architecture patterns, and AI infrastructure
category	ai
tags	["ai-architecture","mlops","system-design","ai-infrastructure","scalability"]
allowed-tools	["Read","Write","Edit"]

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)
         

     ():
        
         model  .models.get(name, []):
             model.version == version:
                model.stage = stage
                 
         

     () -> [ModelMetadata]:
        
         model  .models.get(name, []):
             model.stage == ModelStage.PRODUCTION:
                 model
         

 :
    

     ():
        .features: [, ] = {}
        .feature_groups: [, []] = {}

     ():
        
        .features[name] = {
            : dtype,
            : description,
            : transformation
        }

     ():
        
        .feature_groups[group_name] = feature_names

     () -> :
        
        
         {name: ._fetch_feature(entity_id, name)
                 name  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):
    

     ():
        ().__init__(config)
        .world_size = world_size
        .rank = rank
        .setup_distributed()

     ():
        
        dist.init_process_group(
            backend=,
            world_size=.world_size,
            rank=.rank
        )

     ():
        
         torch.utils.data.distributed  DistributedSampler

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

     ():
        
         torch.nn.parallel  DistributedDataParallel  DDP

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

         epoch  (.config[]):
            .sampler.set_epoch(epoch)

             batch  .dataloader:
                loss = .train_step(model, batch)

                 .rank == :
                    .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}:"] = model

         model


      () -> :
        
        prediction_counter.inc()

        
        cache_key = ._generate_cache_key(model_name, features)
         cache_key  .prediction_cache:
             .prediction_cache[cache_key]

        
        model = .loaded_models.get(model_name)
          model:
            model =  .load_model(model_name)

        
        result =  ._run_inference(model, features)

        
        .prediction_cache[cache_key] = result

         result

      () -> []:
        
        tasks = [.predict(model_name, features)
                 features  batch_features]
          asyncio.gather(*tasks)

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


  ():
      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": [],
                : 
            }

        .metrics[model_key][].append(log.latency_ms)
        .metrics[model_key][] += 

     ():
        
        
        
        

     () -> :
        
        model_metrics = .metrics.get(model_name, {})

        latencies = model_metrics.get(, [])

         {
            : model_metrics.get(, ),
            : np.mean(latencies)  latencies  ,
            : np.percentile(latencies, )  latencies  ,
            : np.percentile(latencies, )  latencies  
        }

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/