菜单
默认使用会先检查来源的 Prompt；你也可以切换为直接命令，或下载本地副本。
决定是否安装前，请先阅读 SKILL.md，以及 SkillsMP 当前展示的配套文件。
用 Codex 或 Claude 帮你安装 复制这段 Prompt，粘贴到 Codex、Claude 或其他助手里，让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt；运行前请先检查来源。
npx skills add https://github.com/pluginagentmarketplace/custom-plugin-machine-learning --skill ml-deployment命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地？可下载 SkillsMP 当前能够提供的文件。
Discover patterns in unlabeled data using clustering, dimensionality reduction, and anomaly detection
Build computer vision solutions - image classification, object detection, and transfer learning
Build and train neural networks with PyTorch - MLPs, CNNs, and training best practices
基于 SOC 职业分类
正在显示 SKILL.md
| name | ml-deployment |
| description | Deploy ML models to production - APIs, containerization, monitoring, and MLOps |
| version | 1.4.0 |
| sasmp_version | 1.4.0 |
| bonded_agent | 07-model-deployment |
| bond_type | PRIMARY_BOND |
| parameters | {"required":[{"name":"model","type":"object","validation":"Trained model with predict method"}],"optional":[{"name":"port","type":"integer","default":8000,"validation":"1024 <= port <= 65535"},{"name":"workers","type":"integer","default":4}]} |
| retry_logic | {"strategy":"exponential_backoff","max_attempts":3,"base_delay_ms":1000} |
| logging | {"level":"info","metrics":["latency_ms","requests_per_second","error_rate"]} |
Take models from development to production.
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
import joblib
app = FastAPI(title="ML Model API")
model = joblib.load('model.pkl')
class PredictRequest(BaseModel):
features: list[float]
class PredictResponse(BaseModel):
prediction: float
@app.post("/predict", response_model=PredictResponse)
async def predict(request: PredictRequest):
X = np.array([request.features])
prediction = model.predict(X)[0]
return PredictResponse(prediction=float(prediction))
@app.get("/health")
async def health():
return {"status": "healthy"}
import torch
import torch.onnx
# Export PyTorch to ONNX
def export_to_onnx(model, sample_input, path='model.onnx'):
model.()
torch.onnx.export(
model,
sample_input,
path,
export_params=,
opset_version=,
input_names=[],
output_names=[],
dynamic_axes={: {: }, : {: }}
)
onnxruntime ort
session = ort.InferenceSession()
input_name = session.get_inputs()[].name
output = session.run(, {input_name: input_data})[]
# Dockerfile
FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: '3.8'
services:
api:
build: .
ports:
- "8000:8000"
environment:
- MODEL_PATH=/models/model.onnx
volumes:
- ./models:/models:ro
restart: unless-stopped
from prometheus_client import Counter, Histogram, start_http_server
# Define metrics
REQUESTS = Counter('model_requests_total', 'Total requests', ['status'])
LATENCY = Histogram('model_latency_seconds', 'Latency in seconds')
@app.post("/predict")
async def predict(request: PredictRequest):
import time
start = time.time()
try:
prediction = model.predict(request.features)
REQUESTS.labels(status='success').inc()
return {"prediction": prediction}
except Exception as e:
REQUESTS.labels(status='error').inc()
raise
finally:
LATENCY.observe(time.time() - start)
import mlflow
# Log model
with mlflow.start_run():
mlflow.log_params({"n_estimators": 100, "max_depth": 10})
mlflow.log_metrics({"accuracy": 0.95, "f1": 0.93})
mlflow.sklearn.log_model(model, "model")
# Load model
model_uri = "runs:/abc123/model"
model = mlflow.sklearn.load_model(model_uri)
import random
class ABTest:
def __init__(self, variants: dict[str, float]):
self.variants = variants # {"A": 0.5, "B": 0.5}
self.results = {v: {"count": 0, "success": 0} for v in variants}
def get_variant(self, user_id: str) -> str:
random.seed(hash(user_id))
r = random.random()
cumulative = 0
for variant, weight in self.variants.items():
cumulative += weight
if r <= cumulative:
return variant
return list(self.variants.keys())[-1]
def record(self, variant: str, success: bool):
self.results[variant]["count"] += 1
if success:
self.results[variant]["success"] += 1
# TODO: Create a FastAPI service that:
# 1. Loads a model on startup
# 2. Has /predict and /health endpoints
# 3. Validates input with Pydantic
# TODO: Containerize your ML service
# Create Dockerfile and docker-compose.yml
import pytest
from fastapi.testclient import TestClient
def test_health_endpoint():
"""Test health check."""
client = TestClient(app)
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_predict_endpoint():
"""Test prediction."""
client = TestClient(app)
response = client.post("/predict", json={"features": [1.0, 2.0, 3.0]})
assert response.status_code == 200
assert "prediction" in response.json()
| Problem | Cause | Solution |
|---|---|---|
| High latency | Model too large | Quantize, use ONNX |
| Memory leaks | Poor cleanup | Implement proper lifecycle |
| API errors | Input validation | Add Pydantic schemas |
| Scaling issues | Blocking I/O | Use async, add workers |
07-model-deploymentcomputer-visionVersion: 1.4.0 | Status: Production Ready