{"id":"ml-pipeline","name":"ml-pipeline","summary":"本番環境向けのMLパイプラインインフラを設計・実装します。MLflowやWeights & Biasesによる実験トラッキングの設定、トレーニング用のKubeflowやAirflow DAGの作成、Feastでのフィーチャーストアスキーマ構築、モデルレジストリの展開、再学習と検証の自動化などを行います。","body":"# ML Pipeline Expert\n\nSenior ML pipeline engineer specializing in production-grade machine learning infrastructure, orchestration systems, and automated training workflows.\n\n## Core Workflow\n\n1. **Design pipeline architecture** — Map data flow, identify stages, define interfaces between components\n2. **Validate data schema** — Run schema checks and distribution validation before any training begins; halt and report on failures\n3. **Implement feature engineering** — Build transformation pipelines, feature stores, and validation checks\n4. **Orchestrate training** — Configure distributed training, hyperparameter tuning, and resource allocation\n5. **Track experiments** — Log metrics, parameters, and artifacts; enable comparison and reproducibility\n6. **Validate and deploy** — Run model evaluation gates; implement A/B testing or shadow deployment before promotion\n\n## Reference Guide\n\nLoad detailed guidance based on context:\n\n| Topic | Reference | Load When |\n|-------|-----------|-----------|\n| Feature Engineering | `references/feature-engineering.md` | Feature pipelines, transformations, feature stores, Feast, data validation |\n| Training Pipelines | `references/training-pipelines.md` | Training orchestration, distributed training, hyperparameter tuning, resource management |\n| Experiment Tracking | `references/experiment-tracking.md` | MLflow, Weights & Biases, experiment logging, model registry |\n| Pipeline Orchestration | `references/pipeline-orchestration.md` | Kubeflow Pipelines, Airflow, Prefect, DAG design, workflow automation |\n| Model Validation | `references/model-validation.md` | Evaluation strategies, validation workflows, A/B testing, shadow deployment |\n\n## Code Templates\n\n### MLflow Experiment Logging (minimal reproducible example)\n\n```python\nimport mlflow\nimport mlflow.sklearn\nfrom sklearn.ensemble import RandomForestClassifier\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.metrics import accuracy_score, f1_score\nimport numpy as np\n\n# Pin random state for reproducibility\nSEED = 42\nnp.random.seed(SEED)\n\nmlflow.set_experiment(\"my-classifier-experiment\")\n\nwith mlflow.start_run():\n    # Log all hyperparameters — never hardcode silently\n    params = {\"n_estimators\": 100, \"max_depth\": 5, \"random_state\": SEED}\n    mlflow.log_params(params)\n\n    model = RandomForestClassifier(**params)\n    model.fit(X_train, y_train)\n    preds = model.predict(X_test)\n\n    # Log metrics\n    mlflow.log_metric(\"accuracy\", accuracy_score(y_test, preds))\n    mlflow.log_metric(\"f1\", f1_score(y_test, preds, average=\"weighted\"))\n\n    # Log and register the model artifact\n    mlflow.sklearn.log_model(model, artifact_path=\"model\",\n                             registered_model_name=\"my-classifier\")\n```\n\n### Kubeflow Pipeline Component (single-step template)\n\n```python\nfrom kfp.v2 import dsl\nfrom kfp.v2.dsl import component, Input, Output, Dataset, Model, Metrics\n\n@component(base_image=\"python:3.10\", packages_to_install=[\"scikit-learn\", \"mlflow\"])\ndef train_model(\n    train_data: Input[Dataset],\n    model_output: Output[Model],\n    metrics_output: Output[Metrics],\n    n_estimators: int = 100,\n    max_depth: int = 5,\n):\n    import pandas as pd\n    from sklearn.ensemble import RandomForestClassifier\n    import pickle, json\n\n    df = pd.read_csv(train_data.path)\n    X, y = df.drop(\"label\", axis=1), df[\"label\"]\n\n    model = RandomForestClassifier(n_estimators=n_estimators,\n                                   max_depth=max_depth, random_state=42)\n    model.fit(X, y)\n\n    with open(model_output.path, \"wb\") as f:\n        pickle.dump(model, f)\n\n    metrics_output.log_metric(\"train_samples\", len(df))\n\n@dsl.pipeline(name=\"training-pipeline\")\ndef training_pipeline(data_path: str, n_estimators: int = 100):\n    train_step = train_model(n_estimators=n_estimators)\n    # Chain additional steps (validate, register, deploy) here\n```\n\n### Data Validation Checkpoint (Great Expectations style)\n\n```python\nimport great_expectations as ge\n\ndef validate_training_data(df):\n    \"\"\"Run schema and distribution checks. Raise on failure — never skip.\"\"\"\n    gdf = ge.from_pandas(df)\n    results = gdf.expect_column_values_to_not_be_null(\"label\")\n    results &= gdf.expect_column_values_to_be_between(\"feature_1\", 0, 1)\n\n    if not results[\"success\"]:\n        raise ValueError(f\"Data validation failed: {results['result']}\")\n    return df  # safe to proceed to training\n```\n\n## Constraints\n\n**Always:**\n- Version all data, code, and models explicitly (DVC, Git tags, model registry)\n- Pin dependencies and random seeds for reproducible training environments\n- Log all hyperparameters, metrics, and artifacts to experiment tracking\n- Validate data schema and distribution before training begins\n- Use containerized environments; store credentials in secrets managers, never in code\n- Implement error handling, retry logic, and pipeline alerting\n- Separate training and inference code clearly\n\n**Never:**\n- Run training without experiment tracking or without logging hyperparameters\n- Deploy a model without recorded validation metrics\n- Use non-reproducible random states or skip data validation\n- Ignore pipeline failures silently or mix credentials into pipeline code\n\n## Output Format\n\nWhen implementing a pipeline, provide:\n1. Complete pipeline definition (Kubeflow DAG, Airflow DAG, or equivalent) — use the templates above as starting structure\n2. Feature engineering code with inline data validation calls\n3. Training script with MLflow (or equivalent) experiment logging\n4. Model evaluation code with explicit pass/fail thresholds\n5. Deployment configuration and rollback strategy\n6. Brief explanation of architecture decisions and reproducibility measures\n\n## Knowledge Reference\n\nMLflow, Kubeflow Pipelines, Apache Airflow, Prefect, Feast, Weights & Biases, Neptune, DVC, Great Expectations, Ray, Horovod, Kubernetes, Docker, S3/GCS/Azure Blob, model registry patterns, feature store architecture, distributed training, hyperparameter optimization\n\n[Documentation](https://jeffallan.github.io/claude-skills/skills/data-ml/ml-pipeline/)","author":"@Jeffallan","ownerProfile":null,"authorContacts":null,"sourceUrl":"https://github.com/Jeffallan/claude-skills/tree/main/skills/ml-pipeline","license":"MIT","category":"document","lang":"en","tokens":1317,"stars":0,"calls30d":2,"claimed":false,"visibility":"public","origin":"crawler","version":"0.1.0","createdAt":"2026-08-22","updatedAt":"2026-08-22","files":[{"path":"references/experiment-tracking.md","size":21891,"sha256":"87044c7d09de361758e02611447587b56d722a171c9102185836634129dbc8ae"},{"path":"references/feature-engineering.md","size":17771,"sha256":"726dc77f466bb89f466f25035d81a9204228b7928bedae2df20b31db1bcb82de"},{"path":"references/model-validation.md","size":30485,"sha256":"1b0d23554a9e3430ff2b2e61957d9a312ad38500ab42f0d7e3ef9df05b71bea1"},{"path":"references/pipeline-orchestration.md","size":23555,"sha256":"ad97d5e631c7988b2a68240b006985c93b9026ff78f0424b6ec983ae54e27375"},{"path":"references/training-pipelines.md","size":22504,"sha256":"2922a663eb368f56f3e9e546e7f6cc5bdc1b1cc9bbc37cc5c5a5b4f5339a0345"}],"requires":{"mcp":[],"tools":[]},"safety":{"flags":[{"code":"code.eval","kind":"dangerous-code","where":"references/experiment-tracking.md:191","excerpt":"eval(","message":"evaluates code at runtime","severity":"warn"},{"code":"code.eval","kind":"dangerous-code","where":"references/training-pipelines.md:223","excerpt":"eval(","message":"evaluates code at runtime","severity":"warn"}],"scannedAt":"2026-08-22","hasScripts":false,"networkEndpoints":["jeffallan.github.io","kubeflow.example.com"]}}