# Data versioning with DVC
import dvc.api

# Track data changes alongside code
data_path = dvc.api.get_url(
    path='data/training/dataset.csv',
    repo='[email protected]:org/ml-project.git',
    rev='v1.2.0'
)

# Data quality checks
from great_expectations.core import ExpectationSuite

suite = ExpectationSuite("training_data")
suite.add_expectation(
    expect_column_values_to_not_be_null(column="target")
)
suite.add_expectation(
    expect_column_values_to_be_between(
        column="feature_1", min_value=0, max_value=100
    )
)

# Experiment tracking with MLflow
import mlflow

with mlflow.start_run():
    mlflow.log_params({
        "learning_rate": 0.001,
        "batch_size": 32,
        "epochs": 100
    })

    model = train_model(X_train, y_train)

    mlflow.log_metrics({
        "accuracy": accuracy,
        "f1_score": f1,
        "auc_roc": auc
    })

    mlflow.sklearn.log_model(model, "model")

# Model validation against baseline
def validate_model(new_model, baseline_model, test_data):
    new_metrics = evaluate(new_model, test_data)
    baseline_metrics = evaluate(baseline_model, test_data)

    # Statistical significance test
    improvement = new_metrics['accuracy'] - baseline_metrics['accuracy']
    p_value = statistical_test(new_model, baseline_model, test_data)

    return {
        'passes': improvement > 0.01 and p_value < 0.05,
        'improvement': improvement,
        'p_value': p_value
    }

# Model registry and deployment
from mlflow.tracking import MlflowClient

client = MlflowClient()

# Register model
model_uri = f"runs:/{run_id}/model"
mv = client.create_model_version(
    name="fraud_detection",
    source=model_uri,
    run_id=run_id
)

# Promote to production after validation
client.transition_model_version_stage(
    name="fraud_detection",
    version=mv.version,
    stage="Production"
)

# Drift detection
from evidently.metrics import DataDriftPreset
from evidently.report import Report

def check_drift(reference_data, production_data):
    report = Report(metrics=[DataDriftPreset()])
    report.run(
        reference_data=reference_data,
        current_data=production_data
    )

    drift_detected = report.as_dict()['metrics'][0]['result']['dataset_drift']

    if drift_detected:
        trigger_retraining_pipeline()

    return drift_detected

# Static threshold (traditional)
def static_alert(cpu_usage):
    return cpu_usage > 80  # Always alerts at 80%

# ML-based dynamic baseline
from sklearn.ensemble import IsolationForest
import numpy as np

class DynamicBaseline:
    def __init__(self, window_size=168):  # 1 week of hourly data
        self.model = IsolationForest(contamination=0.01)
        self.history = []
        self.window_size = window_size

    def update(self, value, hour_of_week):
        # Include temporal features
        features = [value, hour_of_week, np.sin(2*np.pi*hour_of_week/168)]
        self.history.append(features)

        if len(self.history) > self.window_size:
            self.history.pop(0)
            self.model.fit(self.history)

    def is_anomaly(self, value, hour_of_week):
        features = [[value, hour_of_week, np.sin(2*np.pi*hour_of_week/168)]]
        return self.model.predict(features)[0] == -1

# Example: Kubernetes-based self-healing with custom controller
apiVersion: aiops.io/v1
kind: RemediationPolicy
metadata:
  name: memory-pressure-remediation
spec:
  trigger:
    type: Anomaly
    metric: container_memory_usage_bytes
    model: isolation_forest
    sensitivity: 0.95

  diagnosis:
    correlations:
      - metric: container_restarts
      - metric: http_error_rate
      - logs: "OutOfMemoryError"

  actions:
    - type: ScaleUp
      condition: "correlation.http_error_rate > 0.8"
      target:
        deployment: api-server
        replicas: "+2"

    - type: Restart
      condition: "correlation.container_restarts > 3"
      target:
        pod: "selector:app=api-server"

  feedback:
    successMetric: http_error_rate
    successThreshold: 0.01
    learningEnabled: true

class AlertCorrelator:
    def __init__(self, topology_graph, time_window=300):
        self.topology = topology_graph
        self.time_window = time_window
        self.alert_buffer = []

    def process_alert(self, alert):
        self.alert_buffer.append(alert)
        self.cleanup_old_alerts()

        # Group alerts by time proximity
        time_groups = self.group_by_time(self.alert_buffer)

        # Correlate by topology
        for group in time_groups:
            correlated = self.correlate_by_topology(group)

            # Find root cause candidates
            root_causes = self.find_root_causes(correlated)

            # Score by business impact
            scored = self.score_impact(root_causes)

            yield Incident(
                alerts=correlated,
                probable_causes=scored[:3],
                suggested_actions=self.get_runbooks(scored[0])
            )

    def correlate_by_topology(self, alerts):
        # Use graph algorithms to find related services
        clusters = []
        for alert in alerts:
            service = alert.source
            neighbors = self.topology.neighbors(service, depth=2)
            related = [a for a in alerts if a.source in neighbors]
            clusters.append(set(related + [alert]))

        # Merge overlapping clusters
        return self.merge_clusters(clusters)

# Performance comparison (requests per second on single A100)
optimizations = {
    "naive": 10,          # Basic implementation
    "continuous_batching": 50,   # 5x improvement
    "paged_attention": 150,      # 15x improvement (vLLM)
    "quantization_int8": 250,    # 25x improvement
    "quantization_int4": 300,    # 30x improvement
}

# vLLM deployment example
from vllm import LLM, SamplingParams

llm = LLM(
    model="meta-llama/Llama-3-70B-Instruct",
    tensor_parallel_size=4,  # Distribute across 4 GPUs
    quantization="awq",       # 4-bit quantization
    gpu_memory_utilization=0.9
)

sampling_params = SamplingParams(
    temperature=0.7,
    max_tokens=1024,
    top_p=0.95
)

# Handles batching automatically
outputs = llm.generate(prompts, sampling_params)

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticCache:
    def __init__(self, similarity_threshold=0.95):
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
        self.cache = {}  # embedding -> response
        self.threshold = similarity_threshold

    def get(self, query):
        query_embedding = self.encoder.encode(query)

        for cached_embedding, response in self.cache.items():
            similarity = np.dot(query_embedding, cached_embedding)
            if similarity > self.threshold:
                return response
        return None

    def set(self, query, response):
        embedding = tuple(self.encoder.encode(query))
        self.cache[embedding] = response

def detect_prompt_injection(user_input):
    dangerous_patterns = [
        r"ignore (all )?(previous|prior|above) instructions",
        r"disregard (the )?(system|initial) prompt",
        r"you are now",
        r"new instruction:",
        r"```system",
    ]

    for pattern in dangerous_patterns:
        if re.search(pattern, user_input, re.IGNORECASE):
            return True

    # ML-based detection for sophisticated attacks
    injection_score = injection_classifier.predict_proba([user_input])[0][1]
    return injection_score > 0.8

class ModelRouter:
    def __init__(self):
        self.complexity_classifier = load_classifier()
        self.models = {
            "simple": ("gpt-3.5-turbo", 0.002),    # $/1K tokens
            "medium": ("claude-3-haiku", 0.003),
            "complex": ("gpt-4", 0.06)
        }

    def route(self, prompt):
        complexity = self.complexity_classifier.predict([prompt])[0]
        model, cost = self.models[complexity]
        return model

    # 40-60% cost savings by routing simple queries to cheaper models

# prompts/customer_support/v2.1.yaml
metadata:
  name: customer_support_agent
  version: "2.1"
  author: "platform-team"
  created: "2026-01-15"

model:
  provider: anthropic
  name: claude-3-5-sonnet
  temperature: 0.3
  max_tokens: 1024

system_prompt: |
  You are a helpful customer support agent for TechCorp.

  Guidelines:
  - Be concise and professional
  - If you don't know the answer, say so
  - Never make up information about products
  - For billing issues, always offer to escalate

  Available actions:
  - lookup_order(order_id)
  - check_inventory(product_id)
  - create_ticket(category, description)  

user_prompt_template: |
  Customer: {customer_name}
  Membership: {membership_tier}
  Previous interactions: {interaction_count}

  Query: {user_message}  

tests:
  - input:
      user_message: "Where is my order #12345?"
    expected_contains:
      - "lookup_order"
    expected_not_contains:
      - "I don't have access"

  - input:
      user_message: "Your product is terrible"
    expected_tone: "empathetic"
    expected_not_contains:
      - "I agree"

# .github/workflows/prompt-ci.yaml
name: Prompt CI/CD

on:
  push:
    paths:
      - 'prompts/**'

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run prompt tests
        run: |
          python scripts/test_prompts.py prompts/          

      - name: Evaluate quality metrics
        run: |
          python scripts/evaluate_prompts.py \
            --prompts prompts/ \
            --min-relevancy 0.8 \
            --min-faithfulness 0.9          

      - name: Deploy to staging
        if: github.ref == 'refs/heads/main'
        run: |
          python scripts/deploy_prompts.py \
            --env staging \
            --prompts prompts/          

# GPU-enabled Kubernetes deployment for model training
apiVersion: kubeflow.org/v1
kind: PyTorchJob
metadata:
  name: distributed-training
spec:
  pytorchReplicaSpecs:
    Master:
      replicas: 1
      template:
        spec:
          containers:
            - name: pytorch
              image: training-image:v1.0
              resources:
                limits:
                  nvidia.com/gpu: 8
              env:
                - name: NCCL_DEBUG
                  value: "INFO"
    Worker:
      replicas: 3
      template:
        spec:
          containers:
            - name: pytorch
              image: training-image:v1.0
              resources:
                limits:
                  nvidia.com/gpu: 8
          nodeSelector:
            accelerator: nvidia-h100
          tolerations:
            - key: nvidia.com/gpu
              operator: Exists
              effect: NoSchedule

# MIG configuration for inference workloads
apiVersion: v1
kind: ConfigMap
metadata:
  name: mig-config
data:
  config.yaml: |
    version: v1
    mig-configs:
      all-balanced:
        - devices: all
          mig-enabled: true
          mig-devices:
            "3g.40gb": 2  # Two 3-GPU slices with 40GB each    

# Time-slicing configuration
apiVersion: v1
kind: ConfigMap
metadata:
  name: time-slicing-config
data:
  any: |-
    version: v1
    sharing:
      timeSlicing:
        resources:
          - name: nvidia.com/gpu
            replicas: 4  # Allow 4 pods to share each GPU    

# Training with automatic checkpointing for spot instances
import signal

def checkpoint_handler(signum, frame):
    save_checkpoint(model, optimizer, epoch)
    sys.exit(0)

signal.signal(signal.SIGTERM, checkpoint_handler)

# Resume from checkpoint if exists
if checkpoint_exists():
    model, optimizer, start_epoch = load_checkpoint()
else:
    start_epoch = 0

for epoch in range(start_epoch, total_epochs):
    train_epoch(model, data)
    save_checkpoint(model, optimizer, epoch)  # Regular checkpoints

# KNative service with scale-to-zero
apiVersion: serving.knative.dev/v1
kind: Service
metadata:
  name: inference-service
spec:
  template:
    metadata:
      annotations:
        autoscaling.knative.dev/min-scale: "0"
        autoscaling.knative.dev/max-scale: "10"
        autoscaling.knative.dev/scale-down-delay: "5m"
    spec:
      containers:
        - image: inference:v1
          resources:
            limits:
              nvidia.com/gpu: 1

# Export to ONNX for portability
import torch.onnx

torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    opset_version=17,
    input_names=['input'],
    output_names=['output'],
    dynamic_axes={
        'input': {0: 'batch_size'},
        'output': {0: 'batch_size'}
    }
)