# Model Observability > > - **Type:** Skill - **Install:** `agentstack add skill-timwukp-mlops-agent-skills-model-observability` - **Verified:** Yes — security-reviewed for prompt injection and unsafe behavior - **Seller:** [timwukp](https://agentstack.voostack.com/s/timwukp) - **Installs:** 0 - **Category:** [Agent Skills](https://agentstack.voostack.com/c/agent-skills) - **Latest version:** 0.1.0 - **License:** Apache-2.0 - **Upstream author:** [timwukp](https://github.com/timwukp) - **Source:** https://github.com/timwukp/MLOps-agent-skills/tree/main/skills/mlops/model-observability ## Install ```sh agentstack add skill-timwukp-mlops-agent-skills-model-observability ``` Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent. ## About # Model Observability ## Overview ML observability goes beyond monitoring by providing the ability to understand WHY a model is behaving a certain way - through explainability, tracing, logging, and slice analysis. ## When to Use This Skill - Debugging why a model makes certain predictions - Adding explainability to model outputs - Setting up distributed tracing for ML pipelines - Building fairness and bias detection systems - Creating audit trails for compliance ## Three Pillars of ML Observability ``` ML Observability ┌──────────┬──────────┐ │ Metrics │ Logs │ Traces │ │ │ │ accuracy │ predict │ request → │ latency │ inputs │ preprocess → │ drift │ outputs │ inference → │ fairness │ errors │ postprocess → │ │ │ response └──────────┴──────────┘ ``` ## Step-by-Step Instructions ### 1. SHAP Explanations ```python import shap import numpy as np class ModelExplainer: def __init__(self, model, model_type="tree"): self.model = model if model_type == "tree": self.explainer = shap.TreeExplainer(model) elif model_type == "linear": self.explainer = shap.LinearExplainer(model, X_train) else: self.explainer = shap.KernelExplainer(model.predict, shap.sample(X_train, 100)) def explain_single(self, instance): """Explain a single prediction.""" shap_values = self.explainer.shap_values(instance.reshape(1, -1)) return { "base_value": float(self.explainer.expected_value), "prediction": float(self.model.predict(instance.reshape(1, -1))[0]), "feature_contributions": dict(zip( feature_names, [float(v) for v in shap_values[0]] )), "top_positive": sorted( zip(feature_names, shap_values[0]), key=lambda x: x[1], reverse=True )[:5], "top_negative": sorted( zip(feature_names, shap_values[0]), key=lambda x: x[1] )[:5], } def explain_batch(self, X, save_path=None): """Generate SHAP summary for a batch.""" shap_values = self.explainer.shap_values(X) if save_path: shap.summary_plot(shap_values, X, feature_names=feature_names, show=False) import matplotlib.pyplot as plt plt.savefig(save_path, bbox_inches="tight", dpi=150) plt.close() # Feature importance ranking importance = np.abs(shap_values).mean(axis=0) return dict(sorted( zip(feature_names, importance), key=lambda x: x[1], reverse=True )) ``` ### 2. LIME Explanations ```python from lime.lime_tabular import LimeTabularExplainer explainer = LimeTabularExplainer( X_train.values, feature_names=feature_names, class_names=class_names, mode="classification", ) explanation = explainer.explain_instance( instance, model.predict_proba, num_features=10, num_samples=5000, ) # Get feature contributions contributions = explanation.as_list() # [('feature_a > 0.5', 0.15), ('feature_b = 1', -0.08), ...] ``` ### 3. Structured Prediction Logging ```python import json import logging from datetime import datetime from uuid import uuid4 class PredictionLogger: def __init__(self, log_path="predictions.jsonl"): self.logger = logging.getLogger("predictions") handler = logging.FileHandler(log_path) handler.setFormatter(logging.Formatter("%(message)s")) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) def log(self, features, prediction, model_version, metadata=None): record = { "prediction_id": str(uuid4()), "timestamp": datetime.utcnow().isoformat(), "model_version": model_version, "features": features if isinstance(features, dict) else features.tolist(), "prediction": float(prediction) if not isinstance(prediction, list) else prediction, "metadata": metadata or {}, } self.logger.info(json.dumps(record)) return record["prediction_id"] ``` ### 4. OpenTelemetry Tracing ```python from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter # Setup provider = TracerProvider() processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://jaeger:4317")) provider.add_span_processor(processor) trace.set_tracer_provider(provider) tracer = trace.get_tracer("ml-service") def predict_with_tracing(features, model, preprocessor): with tracer.start_as_current_span("prediction_pipeline") as span: span.set_attribute("model.version", "v1.0") span.set_attribute("input.feature_count", len(features)) with tracer.start_as_current_span("preprocessing"): processed = preprocessor.transform(features) span.set_attribute("preprocessing.output_shape", str(processed.shape)) with tracer.start_as_current_span("inference"): prediction = model.predict(processed) span.set_attribute("prediction.value", float(prediction[0])) with tracer.start_as_current_span("postprocessing"): result = postprocess(prediction) return result ``` ### 5. Slice-Based Analysis ```python def slice_analysis(df, prediction_col, target_col, slice_columns): """Analyze model performance across data slices.""" results = [] for col in slice_columns: for slice_value in df[col].unique(): mask = df[col] == slice_value subset = df[mask] if len(subset) 0 else 0 metrics["equalized_odds_gap"] = max( [m["true_positive_rate"] for m in metrics.values() if isinstance(m, dict)] ) - min( [m["true_positive_rate"] for m in metrics.values() if isinstance(m, dict)] ) return metrics ``` ## Best Practices 1. **Log every prediction** with a unique ID for traceability 2. **Explain outlier predictions** automatically (high confidence wrong, low confidence) 3. **Slice by business segments** - not just overall metrics 4. **Use SHAP for global, LIME for local** explanations 5. **Add OpenTelemetry** early - retrofitting tracing is painful 6. **Monitor fairness metrics** alongside performance metrics 7. **Build audit trails** for regulated industries 8. **Set retention policies** for prediction logs 9. **Profile explanations** - SHAP can be slow for large models ## Scripts - `scripts/explain_predictions.py` - SHAP/LIME explainability engine - `scripts/prediction_logger.py` - Structured prediction logging with tracing ## References See [references/REFERENCE.md](references/REFERENCE.md) for platform comparisons. ## Source & license This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code. - **Author:** [timwukp](https://github.com/timwukp) - **Source:** [timwukp/MLOps-agent-skills](https://github.com/timwukp/MLOps-agent-skills) - **License:** Apache-2.0 Install and usage instructions live in the source repository linked above. ## Pricing - **Free** — Free ## Security capabilities Automated source analysis of v0.1.0 — what this tool can access: - **Network access:** no - **Filesystem access:** no - **Shell / process execution:** no - **Environment & secrets:** no - **Dynamic code execution:** no *"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.* ## Versions - **0.1.0** — security scan: passed — Imported from the upstream source. ## Links - Listing page: https://agentstack.voostack.com/l/skill-timwukp-mlops-agent-skills-model-observability - Seller: https://agentstack.voostack.com/s/timwukp - Browse the marketplace: https://agentstack.voostack.com/browse --- Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.