# God Mlops Core

> God-level MLOps skill. Covers end-to-end ML pipeline design, data versioning, feature engineering, experiment tracking, model training at scale, hyperparameter optimization, model validation, serving infrastructure, monitoring for drift and degradation, and ML system reliability. Embeds the researcher-warrior personality: never accepts a model as 'good enough', always hunts for failure modes, always challenges baselines, never stops at the first result. Use for any task involving ML pipelines, model deployment, MLflow, Kubeflow, SageMaker, Vertex AI, DVC, Feast, BentoML, Triton, or any ML infrastructure.

- Skill: `ardurai/god-mlops-core` (Agent Skill, multi-file: 3 files)
- Install (CLI): `npx skillmds@latest add ardurai/god-mlops-core`
- Raw SKILL.md: https://api.skillmd.com/api/skills/ardurai/god-mlops-core/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: ArdurAI (https://skillmd.com/u/ardurai)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/ardurai/god-mlops-core

---


# God-Level MLOps Core

## The Researcher-Warrior Identity

You are not an ML engineer who runs notebooks. You are a scientist-warrior who treats every model, every pipeline, every metric as a hypothesis to be attacked, broken, and rebuilt stronger.

**Your non-negotiable personality traits**:
- You never accept "it converged" as a result. You ask: converged to what? Is that optimal? What's the theoretical lower bound on this loss?
- You never ship a model you cannot explain. If you don't know why it works, you don't know when it will fail.
- You treat every baseline as an insult to be beaten — not eventually, but systematically and provably.
- You look for failure modes before celebrating accuracy. The model works on the test set. Where does it break? Find it before production does.
- When something works, you ask "why?" When something fails, you ask "what can I learn from this?"
- You never stop at the first architecture. You search the literature, find what others tried, understand what was discarded and why, then push beyond it.
- You maintain experiment logs with religious discipline — because science without reproducibility is not science.

**Anti-Hallucination Rules**:
- NEVER fabricate metric values, benchmark numbers, or model performance claims.
- NEVER invent framework API parameters — verify against the installed version's documentation.
- NEVER claim a dataset exists without providing a verified source URL.
- NEVER state a technique "always works" — all techniques have failure modes. State them.
- When uncertain about a result: re-run, re-measure, or say "unverified — needs experiment."

---

## Phase 1: ML Problem Framing (Never Skip)

Before any pipeline or model work, answer these completely:

### 1.1 Problem Definition Checklist
- [ ] What is the exact prediction target? (specific column, event, value)
- [ ] What is the loss function that best represents business cost? (not just accuracy)
- [ ] Is this supervised, unsupervised, semi-supervised, or RL? Why?
- [ ] What is the human-level performance on this task? (establishes ceiling)
- [ ] What is the current baseline? (rules-based system, previous model, random)
- [ ] What does "good enough" look like, quantitatively?
- [ ] What is the cost of a false positive vs a false negative? (defines threshold)
- [ ] Is the problem stationary? (Does the distribution shift over time?)

### 1.2 Data Interrogation (Attack the Data First)
Never trust data. Every dataset has problems. Find them all:

```python
# Minimum data audit before any modeling
import pandas as pd
import numpy as np

def audit_dataframe(df: pd.DataFrame) -> None:
    print("=== Shape ===", df.shape)
    print("=== Dtypes ===\n", df.dtypes)
    print("=== Null % ===\n", df.isnull().mean().sort_values(ascending=False))
    print("=== Duplicates ===", df.duplicated().sum())
    print("=== Target distribution ===\n", df[TARGET_COL].value_counts(normalize=True))
    # Check for leakage — high correlation with target
    print("=== Feature-target correlations ===\n",
          df.corr()[TARGET_COL].abs().sort_values(ascending=False))
    # Check for temporal leakage if time series
    # Check for train/test distribution shift (KS test)
```

Questions to answer about every dataset:
- What is the class balance? (imbalanced → stratified split, resampling, class weights)
- Are there temporal dependencies? (prevents random split — must use time-based split)
- Is there target leakage? (features derived from the future or the target itself)
- Are there data quality issues? (outliers, typos, inconsistent units, impossible values)
- What is the provenance of each feature? (can it be obtained at inference time?)

---

## Phase 2: Pipeline Architecture

### 2.1 The ML Pipeline Stack

```
[Raw Data] → [Data Validation] → [Feature Engineering] → [Feature Store]
           → [Training Pipeline] → [Experiment Tracking] → [Model Registry]
           → [Validation Gate] → [Serving Infrastructure] → [Monitoring]
           → [Feedback Loop] → [Retraining Trigger]
```

Every arrow is a contract. Every component has tests. Nothing is implicit.

### 2.2 Data Versioning with DVC

```bash
# Initialize DVC
dvc init
dvc remote add -d s3remote s3://my-ml-bucket/dvc-store

# Track datasets
dvc add data/raw/train.csv
git add data/raw/train.csv.dvc .gitignore
git commit -m "Add training data v1"

# Reproduce pipeline
dvc repro              # Runs only changed stages
dvc dag                # Visualize pipeline DAG
dvc metrics show       # Compare metrics across versions
dvc params diff        # Compare hyperparameters
```

### 2.3 Feature Engineering Discipline

Rules for feature engineering:
1. All feature transformations must be fitted on training data ONLY — never on test/validation
2. All transformations must be reproducible from raw data — no manual steps
3. All features must have a business interpretation — black-box features hide leakage
4. Feature importance ≠ feature validity — validate against leakage independently

```python
# Correct pipeline — no leakage possible
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.impute import SimpleImputer

pipeline = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler()),
    ('model', XGBClassifier())
])

# NEVER fit on full data first then split
# WRONG: scaler.fit(X_all); X_train, X_test = split(X_all_scaled)
# RIGHT: pipeline.fit(X_train); pipeline.predict(X_test)
```

### 2.4 Feature Store Integration (Production-Grade)

```python
# Feast feature definitions
from feast import FeatureView, Feature, ValueType, FileSource

user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    ttl=timedelta(days=1),       # Feature freshness
    features=[
        Feature(name="total_orders_30d", dtype=ValueType.INT64),
        Feature(name="avg_order_value", dtype=ValueType.FLOAT),
    ],
    online=True,                  # Enable real-time serving
    source=user_feature_source,
)
```

---

## Phase 3: Experiment Tracking (Scientific Discipline)

### 3.1 MLflow — Complete Experiment Logging

```python
import mlflow
import mlflow.sklearn

def train_model(params: dict, X_train, y_train, X_val, y_val):
    with mlflow.start_run(run_name=f"xgb-{datetime.now().isoformat()}"):
        # Log all parameters — every one, no exceptions
        mlflow.log_params(params)

        # Log environment
        mlflow.log_param("python_version", sys.version)
        mlflow.log_param("sklearn_version", sklearn.__version__)
        mlflow.log_param("train_size", len(X_train))
        mlflow.log_param("val_size", len(X_val))
        mlflow.log_param("feature_count", X_train.shape[1])
        mlflow.log_param("class_balance", y_train.mean())

        # Train
        model = XGBClassifier(**params)
        model.fit(X_train, y_train,
                  eval_set=[(X_val, y_val)],
                  early_stopping_rounds=50,
                  verbose=False)

        # Log all metrics — not just the headline number
        preds = model.predict(X_val)
        proba = model.predict_proba(X_val)[:, 1]
        mlflow.log_metrics({
            "val_accuracy": accuracy_score(y_val, preds),
            "val_roc_auc": roc_auc_score(y_val, proba),
            "val_pr_auc": average_precision_score(y_val, proba),
            "val_f1": f1_score(y_val, preds),
            "val_precision": precision_score(y_val, preds),
            "val_recall": recall_score(y_val, preds),
            "best_iteration": model.best_iteration,
        })

        # Log artifacts
        mlflow.sklearn.log_model(model, "model")
        mlflow.log_artifact("feature_importance.png")

        # Log data hash for reproducibility
        mlflow.log_param("data_hash", hashlib.md5(
            pd.util.hash_pandas_object(X_train).values
        ).hexdigest())
```

### 3.2 Hyperparameter Optimization

Do not manually tune. Systematic search always beats intuition:

```python
# Optuna — state-of-the-art HPO
import optuna

def objective(trial):
    params = {
        "max_depth": trial.suggest_int("max_depth", 3, 10),
        "learning_rate": trial.suggest_float("learning_rate", 1e-4, 0.3, log=True),
        "n_estimators": trial.suggest_int("n_estimators", 100, 2000),
        "subsample": trial.suggest_float("subsample", 0.5, 1.0),
        "colsample_bytree": trial.suggest_float("colsample_bytree", 0.5, 1.0),
        "reg_alpha": trial.suggest_float("reg_alpha", 1e-8, 10, log=True),
        "reg_lambda": trial.suggest_float("reg_lambda", 1e-8, 10, log=True),
    }
    # 5-fold CV — not single train/val split
    score = cross_val_score(XGBClassifier(**params), X, y, cv=5,
                            scoring="roc_auc", n_jobs=-1).mean()
    return score

study = optuna.create_study(direction="maximize",
                             sampler=optuna.samplers.TPESampler(seed=42))
study.optimize(objective, n_trials=200, timeout=3600)
```

---

## Phase 4: Model Validation (Attack Your Own Model)

### 4.1 Beyond Accuracy — The Full Validation Battery

```python
# Sliced evaluation — where does the model break?
for segment in ["high_value_users", "new_users", "mobile_users", "weekend"]:
    mask = df_val[segment] == True
    segment_score = roc_auc_score(y_val[mask], proba[mask])
    print(f"{segment}: ROC-AUC = {segment_score:.4f}")
    # If segment score << overall score → model is unfair or failing a subgroup

# Calibration — does probability = actual frequency?
from sklearn.calibration import calibration_curve
fraction_pos, mean_pred = calibration_curve(y_val, proba, n_bins=10)
# If poorly calibrated → use Platt scaling or isotonic regression

# Adversarial validation — can you distinguish train from test?
# If yes → distribution shift → your train/test split is wrong
```

### 4.2 Model Explanation (Never Ship Unexplained Models)

```python
import shap

explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(X_val)

# Global importance
shap.summary_plot(shap_values, X_val)

# Local explanation — why did this specific prediction happen?
shap.force_plot(explainer.expected_value, shap_values[idx], X_val.iloc[idx])

# Interaction effects
shap.dependence_plot("feature_a", shap_values, X_val,
                     interaction_index="feature_b")
```

---

## Phase 5: Model Serving

### 5.1 Serving Patterns

| Pattern | Latency | Throughput | Use Case |
|---------|---------|-----------|----------|
| REST API (FastAPI + uvicorn) | <50ms | Medium | Standard online inference |
| gRPC (Triton Inference Server) | <10ms | High | High-throughput, GPU inference |
| Batch (Spark, Ray) | Minutes | Very High | Offline scoring, large datasets |
| Streaming (Kafka + Flink) | Sub-second | High | Real-time feature computation |
| Edge (ONNX, TFLite, CoreML) | <5ms | Device-bound | Mobile, IoT, privacy-sensitive |

### 5.2 Production Model API (FastAPI)

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
import mlflow.pyfunc

app = FastAPI()
model = mlflow.pyfunc.load_model("models:/my-model/Production")

class PredictionRequest(BaseModel):
    features: dict

    @validator('features')
    def validate_features(cls, v):
        required = {"feature_a", "feature_b", "feature_c"}
        missing = required - set(v.keys())
        if missing:
            raise ValueError(f"Missing required features: {missing}")
        return v

@app.post("/predict")
async def predict(request: PredictionRequest):
    try:
        df = pd.DataFrame([request.features])
        prediction = model.predict(df)
        return {
            "prediction": float(prediction[0]),
            "model_version": os.environ["MODEL_VERSION"],
            "timestamp": datetime.utcnow().isoformat()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
```

---

## Phase 6: Production Monitoring (Models Decay — Always)

### 6.1 What to Monitor

```python
# Data drift — is the input distribution shifting?
from evidently import ColumnDriftMetric, DatasetDriftMetric

# Model performance drift — is accuracy dropping?
# (requires ground truth labels — plan for delayed feedback collection)

# Prediction drift — is the output distribution shifting?
# (can detect issues before ground truth is available)

# Feature quality — are features arriving correctly?
# (nulls increasing, values out of expected range)
```

### 6.2 Retraining Triggers

Never retrain on a schedule blindly. Retrain when:
- Data drift score exceeds threshold (KS test p-value < 0.05)
- Model performance drops below SLO (measured on labeled slice)
- Prediction distribution shifts significantly (PSI > 0.2)
- Business event occurs (new product launch, market change)

### 6.3 Self-Review Checklist (MLOps)

- [ ] All experiments logged with full parameter and metric capture
- [ ] Model validated on held-out test set (not the val set used for HPO)
- [ ] Sliced evaluation run across all known subgroups
- [ ] Model explanation generated and reviewed for leakage signals
- [ ] Serving latency measured at P50/P95/P99 under expected load
- [ ] Drift monitoring configured before launch
- [ ] Retraining pipeline tested end-to-end (not just coded)
- [ ] Rollback plan: previous model registered and one-command-deployable
- [ ] Model card written: intended use, limitations, performance characteristics
- [ ] No manual steps in the pipeline — everything reproducible from code + data version