Chapter 8: Enterprise Integration

Production Deployment, MLOps, and Governance

弘益人間 · Benefit All Humanity

8.1 Enterprise AI Deployment Landscape

Enterprise AI deployment requires robust infrastructure, monitoring, governance, and compliance. WIA-AI-014 Neural Network Format standard enables seamless integration with enterprise MLOps platforms, ensuring models are production-ready, secure, and governable at scale.

8.2 MLOps Pipeline Architecture

Complete ML pipeline from development to production:

MLOps Pipeline with WIA-AI-014:

1. Development
   ├── Model Training (PyTorch/TensorFlow)
   ├── Experiment Tracking (MLflow/W&B)
   └── Version Control (Git + DVC)

2. Validation
   ├── Export to WIA-AI-014 Format
   ├── Model Validation Tests
   ├── Performance Benchmarking
   └── Security Scanning

3. Registration
   ├── Model Registry (MLflow/Custom)
   ├── Metadata Enrichment
   ├── Governance Approval
   └── Artifact Storage (S3/GCS)

4. Deployment
   ├── A/B Testing Infrastructure
   ├── Blue-Green Deployment
   ├── Canary Releases
   └── Multi-Cloud Support

5. Monitoring
   ├── Performance Metrics
   ├── Data Drift Detection
   ├── Model Drift Detection
   └── Alerting & Paging

6. Maintenance
   ├── Model Retraining
   ├── Rollback Procedures
   └── Continuous Improvement

8.3 Model Serving Infrastructure

Kubernetes Deployment

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-server
  labels:
    app: ml-model
    version: v2.1.0
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ml-model
  template:
    metadata:
      labels:
        app: ml-model
        version: v2.1.0
    spec:
      containers:
      - name: model-server
        image: company/model-server:latest
        ports:
        - containerPort: 8080
        env:
        - name: MODEL_PATH
          value: "/models/resnet50_v2.1.0.onnx"
        - name: BATCH_SIZE
          value: "32"
        resources:
          limits:
            memory: "4Gi"
            cpu: "2"
            nvidia.com/gpu: "1"
          requests:
            memory: "2Gi"
            cpu: "1"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 15
          periodSeconds: 5
        volumeMounts:
        - name: model-storage
          mountPath: /models
      volumes:
      - name: model-storage
        persistentVolumeClaim:
          claimName: model-pvc

---
apiVersion: v1
kind: Service
metadata:
  name: model-service
spec:
  selector:
    app: ml-model
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8080
  type: LoadBalancer

Multi-Model Server

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import onnxruntime as ort
from typing import Dict, List
import numpy as np

app = FastAPI()

class ModelRegistry:
    def __init__(self):
        self.models: Dict[str, ort.InferenceSession] = {}

    def load_model(self, name: str, path: str):
        """Load model into registry"""
        session = ort.InferenceSession(path)
        self.models[name] = session
        return True

    def get_model(self, name: str):
        """Retrieve model from registry"""
        if name not in self.models:
            raise HTTPException(status_code=404, detail="Model not found")
        return self.models[name]

    def list_models(self) -> List[str]:
        """List all loaded models"""
        return list(self.models.keys())

registry = ModelRegistry()

# Load models on startup
@app.on_event("startup")
async def load_models():
    registry.load_model("resnet50_v2", "/models/resnet50_v2.1.0.onnx")
    registry.load_model("bert_base", "/models/bert_base_v1.0.0.onnx")

class PredictionRequest(BaseModel):
    model_name: str
    input_data: List[List[float]]

class PredictionResponse(BaseModel):
    model_name: str
    predictions: List[float]
    latency_ms: float

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    import time

    start = time.perf_counter()

    # Get model
    model = registry.get_model(request.model_name)

    # Prepare input
    input_array = np.array(request.input_data, dtype=np.float32)
    input_name = model.get_inputs()[0].name

    # Run inference
    outputs = model.run(None, {input_name: input_array})

    latency = (time.perf_counter() - start) * 1000

    return PredictionResponse(
        model_name=request.model_name,
        predictions=outputs[0].tolist(),
        latency_ms=latency
    )

@app.get("/models")
async def list_models():
    return {"models": registry.list_models()}

@app.get("/health")
async def health():
    return {"status": "healthy"}

# Run: uvicorn server:app --host 0.0.0.0 --port 8080

8.4 A/B Testing and Canary Deployments

import random
from enum import Enum

class DeploymentStrategy(Enum):
    BLUE_GREEN = "blue_green"
    CANARY = "canary"
    AB_TEST = "ab_test"

class ModelRouter:
    def __init__(self):
        self.models = {}
        self.traffic_split = {}

    def register_model(self, name: str, version: str, session, weight: float = 0.0):
        """Register model variant with traffic weight"""
        key = f"{name}:{version}"
        self.models[key] = session
        self.traffic_split[key] = weight

    def route_request(self, model_name: str):
        """Route request to appropriate model version"""
        # Get all versions of this model
        versions = {k: v for k, v in self.traffic_split.items()
                   if k.startswith(f"{model_name}:")}

        if not versions:
            raise ValueError(f"No versions found for {model_name}")

        # Weighted random selection
        total_weight = sum(versions.values())
        rand = random.uniform(0, total_weight)

        cumsum = 0
        for version_key, weight in versions.items():
            cumsum += weight
            if rand <= cumsum:
                return self.models[version_key]

# Usage
router = ModelRouter()

# Blue-Green: 100% on v2, keep v1 warm
router.register_model("resnet50", "v1", model_v1, weight=0.0)
router.register_model("resnet50", "v2", model_v2, weight=1.0)

# Canary: 95% v1, 5% v2
router.register_model("bert", "v1", bert_v1, weight=0.95)
router.register_model("bert", "v2", bert_v2, weight=0.05)

# A/B Test: 50/50 split
router.register_model("gpt", "variant_a", gpt_a, weight=0.5)
router.register_model("gpt", "variant_b", gpt_b, weight=0.5)

# Route request
model = router.route_request("resnet50")
output = model.run(None, input_data)

8.5 Monitoring and Observability

Prometheus Metrics

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Define metrics
prediction_requests = Counter(
    'model_prediction_requests_total',
    'Total number of prediction requests',
    ['model_name', 'version', 'status']
)

prediction_latency = Histogram(
    'model_prediction_latency_seconds',
    'Prediction latency in seconds',
    ['model_name', 'version'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]
)

model_memory_usage = Gauge(
    'model_memory_usage_bytes',
    'Model memory usage in bytes',
    ['model_name', 'version']
)

active_requests = Gauge(
    'model_active_requests',
    'Number of active requests',
    ['model_name', 'version']
)

class MonitoredModel:
    def __init__(self, model, name, version):
        self.model = model
        self.name = name
        self.version = version

    def predict(self, input_data):
        # Increment active requests
        active_requests.labels(
            model_name=self.name,
            version=self.version
        ).inc()

        start = time.perf_counter()

        try:
            # Run inference
            output = self.model.run(None, input_data)

            # Record success
            prediction_requests.labels(
                model_name=self.name,
                version=self.version,
                status='success'
            ).inc()

            return output

        except Exception as e:
            # Record failure
            prediction_requests.labels(
                model_name=self.name,
                version=self.version,
                status='error'
            ).inc()
            raise

        finally:
            # Record latency
            latency = time.perf_counter() - start
            prediction_latency.labels(
                model_name=self.name,
                version=self.version
            ).observe(latency)

            # Decrement active requests
            active_requests.labels(
                model_name=self.name,
                version=self.version
            ).dec()

# Start metrics server
start_http_server(9090)

# Use monitored model
monitored = MonitoredModel(onnx_session, "resnet50", "v2.1.0")
output = monitored.predict(input_data)

Data Drift Detection

import numpy as np
from scipy import stats

class DataDriftDetector:
    def __init__(self, reference_data, significance=0.05):
        self.reference_data = reference_data
        self.significance = significance
        self.reference_stats = self.compute_stats(reference_data)

    def compute_stats(self, data):
        return {
            'mean': np.mean(data, axis=0),
            'std': np.std(data, axis=0),
            'min': np.min(data, axis=0),
            'max': np.max(data, axis=0)
        }

    def detect_drift_ks_test(self, current_data):
        """Kolmogorov-Smirnov test for distribution shift"""
        drifted_features = []

        for i in range(current_data.shape[1]):
            statistic, pvalue = stats.ks_2samp(
                self.reference_data[:, i],
                current_data[:, i]
            )

            if pvalue < self.significance:
                drifted_features.append({
                    'feature_index': i,
                    'statistic': statistic,
                    'pvalue': pvalue
                })

        return {
            'drifted': len(drifted_features) > 0,
            'num_drifted_features': len(drifted_features),
            'details': drifted_features
        }

    def detect_drift_psi(self, current_data, num_bins=10):
        """Population Stability Index"""
        psi_values = []

        for i in range(current_data.shape[1]):
            ref_counts, bin_edges = np.histogram(
                self.reference_data[:, i],
                bins=num_bins
            )
            curr_counts, _ = np.histogram(
                current_data[:, i],
                bins=bin_edges
            )

            # Normalize
            ref_pct = ref_counts / len(self.reference_data)
            curr_pct = curr_counts / len(current_data)

            # Avoid division by zero
            ref_pct = np.where(ref_pct == 0, 0.0001, ref_pct)
            curr_pct = np.where(curr_pct == 0, 0.0001, curr_pct)

            # Calculate PSI
            psi = np.sum((curr_pct - ref_pct) * np.log(curr_pct / ref_pct))
            psi_values.append(psi)

        avg_psi = np.mean(psi_values)

        # PSI interpretation: <0.1 stable, 0.1-0.2 moderate, >0.2 significant
        return {
            'psi': avg_psi,
            'status': 'stable' if avg_psi < 0.1 else 'moderate' if avg_psi < 0.2 else 'significant',
            'feature_psi': psi_values
        }

# Usage
detector = DataDriftDetector(reference_data=training_data)

# Check production data periodically
drift_result = detector.detect_drift_ks_test(production_data)
if drift_result['drifted']:
    print(f"⚠️  Data drift detected in {drift_result['num_drifted_features']} features")
    # Trigger retraining or alerts

8.6 Security and Compliance

Model Encryption

from cryptography.fernet import Fernet
import hashlib

class SecureModelStorage:
    def __init__(self, encryption_key=None):
        if encryption_key is None:
            encryption_key = Fernet.generate_key()
        self.cipher = Fernet(encryption_key)
        self.key = encryption_key

    def encrypt_model(self, model_path, output_path):
        """Encrypt model file"""
        with open(model_path, 'rb') as f:
            model_data = f.read()

        encrypted_data = self.cipher.encrypt(model_data)

        with open(output_path, 'wb') as f:
            f.write(encrypted_data)

        # Compute hash for integrity
        model_hash = hashlib.sha256(model_data).hexdigest()
        return model_hash

    def decrypt_model(self, encrypted_path, output_path):
        """Decrypt model file"""
        with open(encrypted_path, 'rb') as f:
            encrypted_data = f.read()

        decrypted_data = self.cipher.decrypt(encrypted_data)

        with open(output_path, 'wb') as f:
            f.write(decrypted_data)

        return True

# Usage
secure_storage = SecureModelStorage()

# Encrypt before storage
model_hash = secure_storage.encrypt_model(
    'model.onnx',
    'encrypted_model.enc'
)

# Decrypt for loading
secure_storage.decrypt_model(
    'encrypted_model.enc',
    'decrypted_model.onnx'
)

Access Control

from functools import wraps
from fastapi import Header, HTTPException
import jwt

SECRET_KEY = "your-secret-key"

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

def require_role(required_role: str):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, authorization: str = Header(None), **kwargs):
            if not authorization:
                raise HTTPException(status_code=401, detail="No token provided")

            token = authorization.replace("Bearer ", "")
            payload = verify_token(token)

            if payload.get("role") != required_role:
                raise HTTPException(status_code=403, detail="Insufficient permissions")

            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Protected endpoints
@app.post("/models/deploy")
@require_role("admin")
async def deploy_model(model_name: str):
    # Only admins can deploy models
    pass

@app.post("/models/predict")
@require_role("user")
async def predict(request: PredictionRequest):
    # Regular users can make predictions
    pass

8.7 Multi-Cloud Deployment

# Terraform configuration for multi-cloud deployment
# AWS
resource "aws_ecs_service" "model_service" {
  name            = "model-service"
  cluster         = aws_ecs_cluster.main.id
  task_definition = aws_ecs_task_definition.model.arn
  desired_count   = 3

  load_balancer {
    target_group_arn = aws_lb_target_group.model.arn
    container_name   = "model-server"
    container_port   = 8080
  }
}

# GCP
resource "google_cloud_run_service" "model_service" {
  name     = "model-service"
  location = "us-central1"

  template {
    spec {
      containers {
        image = "gcr.io/project/model-server:latest"
        resources {
          limits = {
            cpu    = "2"
            memory = "4Gi"
          }
        }
      }
    }
  }
}

# Azure
resource "azurerm_container_group" "model_service" {
  name                = "model-service"
  location            = azurerm_resource_group.main.location
  resource_group_name = azurerm_resource_group.main.name
  os_type             = "Linux"

  container {
    name   = "model-server"
    image  = "registry/model-server:latest"
    cpu    = "2"
    memory = "4"

    ports {
      port     = 8080
      protocol = "TCP"
    }
  }
}

8.8 Cost Optimization

class CostOptimizer:
    def __init__(self):
        self.request_costs = {
            'cpu': 0.0001,    # per request
            'gpu': 0.001,     # per request
            'memory_gb_hour': 0.01
        }

    def estimate_monthly_cost(self, requests_per_day, avg_latency_ms, memory_gb, use_gpu=False):
        """Estimate monthly infrastructure cost"""
        monthly_requests = requests_per_day * 30
        compute_type = 'gpu' if use_gpu else 'cpu'

        # Compute cost
        compute_cost = monthly_requests * self.request_costs[compute_type]

        # Memory cost (assuming always-on)
        memory_hours = 24 * 30  # hours per month
        memory_cost = memory_gb * memory_hours * self.request_costs['memory_gb_hour']

        total_cost = compute_cost + memory_cost

        return {
            'compute_cost': compute_cost,
            'memory_cost': memory_cost,
            'total_cost': total_cost,
            'cost_per_request': total_cost / monthly_requests
        }

    def recommend_optimization(self, current_cost):
        """Provide cost optimization recommendations"""
        recommendations = []

        if current_cost['compute_cost'] > current_cost['memory_cost'] * 2:
            recommendations.append({
                'type': 'batching',
                'description': 'Implement request batching to reduce per-request overhead',
                'estimated_savings': current_cost['compute_cost'] * 0.3
            })

        if current_cost['memory_cost'] > 100:
            recommendations.append({
                'type': 'quantization',
                'description': 'Apply INT8 quantization to reduce memory footprint',
                'estimated_savings': current_cost['memory_cost'] * 0.5
            })

        return recommendations

# Usage
optimizer = CostOptimizer()
cost = optimizer.estimate_monthly_cost(
    requests_per_day=1000000,
    avg_latency_ms=10,
    memory_gb=4,
    use_gpu=True
)
print(f"Monthly cost: ${cost['total_cost']:.2f}")
print(f"Cost per request: ${cost['cost_per_request']:.6f}")

recommendations = optimizer.recommend_optimization(cost)
for rec in recommendations:
    print(f"\n{rec['type']}: {rec['description']}")
    print(f"Potential savings: ${rec['estimated_savings']:.2f}/month")

8.9 Disaster Recovery

8.10 WIA-AI-014 Enterprise Best Practices