mcpField
M
model-deployment

model-deployment

github.com/seb1n云与 DevOps234114 次安装12 周前更新
代码仓库

关于 model-deployment

这项技能指导 AI Agent 将训练好的机器学习模型封装为推理服务并部署到生产环境，涵盖 API、容器、编排或无服务器部署及运行检查。

主要功能

  1. 01模型序列化与打包
  2. 02FastAPI 或 Flask 服务 API
  3. 03Docker 容器配置
  4. 04Kubernetes 与无服务器部署
  5. 05健康检查、日志和监控

适用场景

  1. 01将训练模型部署为 REST 接口
  2. 02通过 Kubernetes 扩展推理服务
  3. 03发布更新后的模型版本
  4. 04检查生产推理的健康状态与指标

安装

复制下面的命令，在终端中运行。

$npx skills add https://github.com/seb1n/awesome-ai-agent-skills --skill model-deployment

安装后在 Claude Code 中输入/model-deployment确认已加载。

详细介绍

模型部署

此技能可让 AI Agent 将训练好的机器学习模型部署到生产环境。它涵盖将模型打包为 FastAPI 或 Flask 服务 API、使用 Docker 容器化、通过 Kubernetes 编排，以及部署到无服务器平台。Agent 会处理模型版本管理、健康检查、输入验证、日志记录和监控，确保生产环境中的推理可靠且可扩展。

工作流

  1. 序列化并打包模型： 将训练好的模型导出为 ONNX、TorchScript、SavedModel 或 joblib pickle 等可移植格式。将模型文件与其预处理流程及所需的配置文件一同打包，确保推理过程完整自包含。

  2. 构建服务 API： 使用 FastAPI 或 Flask 创建 REST API，在启动时加载模型并提供预测端点。添加健康检查端点、包含输入验证的请求/响应模式（Pydantic 模型）、结构化日志记录，以及可返回有意义 HTTP 状态码的错误处理。

  3. 使用 Docker 容器化： 编写 Dockerfile，从固定版本的 requirements.txt 安装依赖，复制模型文件和服务代码，并将 API 服务设为入口点。使用多阶段构建来尽量减小镜像大小，并避免包含仅用于训练的依赖。

  4. 配置编排和扩缩容： 定义 Kubernetes Deployment 和 Service 清单（或适用于目标平台的等效配置），设置资源请求/限制、指向健康检查端点的就绪和存活探针，以及基于 CPU、内存或请求延迟等自定义指标进行扩缩容的 Horizontal Pod Autoscaler。

  5. 部署并验证： 将容器镜像推送到镜像仓库，应用 Kubernetes 清单或部署到无服务器平台，然后针对线上端点运行冒烟测试。使用一组已知输入验证响应是否符合预期输出。

  6. 监控并迭代： 集成 Prometheus 和 Grafana 等监控工具，跟踪请求延迟、错误率、吞吐量以及预测分布漂移等模型专属指标。为异常设置告警，并建立使用蓝绿或金丝雀策略部署更新模型版本的工作流。

支持的技术

  • API 框架： FastAPI、Flask、TorchServe、TensorFlow Serving、Triton Inference Server
  • 容器化： Docker、Podman
  • 编排： Kubernetes、Docker Compose、AWS ECS、Google Cloud Run
  • 无服务器： AWS Lambda、Google Cloud Functions、Azure Functions
  • 监控： Prometheus、Grafana、Datadog、AWS CloudWatch
  • 模型注册表： MLflow Model Registry、AWS SageMaker Model Registry、Weights & Biases

使用方法

向 Agent 提供训练好的模型文件、其依赖项以及目标部署环境（本地 Docker、Kubernetes 集群或无服务器平台）。Agent 会生成所有必要的服务代码、容器配置和部署清单，然后引导你完成部署流程。

示例

示例 1：使用 FastAPI 部署模型
# app.py
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from contextlib import asynccontextmanager
from typing import List

model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global model
    model = joblib.load("model.pkl")
    yield

app = FastAPI(title="ML Model API", version="1.0.0", lifespan=lifespan)

class PredictionRequest(BaseModel):
    features: List[float]

    @validator("features")
    def validate_features(cls, v):
        if len(v) != 4:
            raise ValueError("Expected exactly 4 features")
        return v

class PredictionResponse(BaseModel):
    prediction: int
    probability: List[float]

@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": model is not None}

@app.post("/predict", response_model=PredictionResponse)
def predict(request: PredictionRequest):
    try:
        features = np.array(request.features).reshape(1, -1)
        prediction = int(model.predict(features)[0])
        probability = model.predict_proba(features)[0].tolist()
        return PredictionResponse(prediction=prediction, probability=probability)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
示例 2：Docker + Kubernetes 部署

Dockerfile：

FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /usr/local/lib/python3.11/site-packages /usr/local/lib/python3.11/site-packages
COPY --from=builder /usr/local/bin/uvicorn /usr/local/bin/uvicorn
COPY app.py model.pkl ./
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

k8s-deployment.yaml：

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ml-model-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ml-model-api
  template:
    metadata:
      labels:
        app: ml-model-api
    spec:
      containers:
        - name: api
          image: registry.example.com/ml-model-api:v1.0.0
          ports:
            - containerPort: 8000
          resources:
            requests: { cpu: "250m", memory: "512Mi" }
            limits: { cpu: "1000m", memory: "1Gi" }
          readinessProbe:
            httpGet: { path: /health, port: 8000 }
            initialDelaySeconds: 10
            periodSeconds: 5
          livenessProbe:
            httpGet: { path: /health, port: 8000 }
            initialDelaySeconds: 15
            periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
  name: ml-model-api
spec:
  selector:
    app: ml-model-api
  ports:
    - port: 80
      targetPort: 8000
  type: LoadBalancer

最佳实践

  • 固定所有依赖项的版本，并使用确定性的 Docker 构建，以确保在不同环境中的可复现性。
  • 将模型制品与代码分离，这样无需重建整个容器镜像即可更新模型。使用模型注册表或采用版本化路径的云存储。
  • 使用 Pydantic 或 JSON Schema 实施输入验证，在格式错误的请求到达模型并产生令人困惑的错误之前将其拒绝。
  • 在 Kubernetes 中使用就绪探针，防止流量到达尚未完成模型加载的 Pod；大型模型的加载可能需要很长时间。
  • 发布新模型版本时采用金丝雀部署——将一小部分流量路由到新版本，并在全面发布前比较各项指标。
  • 记录预测结果和输入（对 PII 进行脱敏），以便在生产环境中调试、审计并检测数据漂移。

边界情况

  • 大型模型文件（> 1 GB）： 避免将它们打包进 Docker 镜像。应在启动时从云存储（S3、GCS）下载，或挂载持久卷。如果模型初始化耗时较长，请使用延迟加载。
  • 无服务器环境中的冷启动延迟： 无服务器函数可能需要 10-30 秒来加载大型模型。可通过预置并发（AWS Lambda）、最小实例数（Cloud Run）或使用 ONNX Runtime 等优化格式来缓解。
  • 推理时预处理不一致： 训练时使用的预处理流水线必须与推理时运行的完全一致。请序列化完整流水线（例如，使用 scikit-learn Pipeline + joblib），不要单独重新实现转换。
  • 优雅关闭和正在处理的请求： 处理 SIGTERM 信号，在关闭前完成正在处理的请求。配置 Kubernetes 的 terminationGracePeriodSeconds，为待处理请求留出足够的完成时间。
  • GPU 与 CPU 推理不匹配： 在 GPU 上训练的模型部署到仅支持 CPU 的环境时可能会失败。加载时明确将模型张量映射到 CPU（torch.load(path, map_location="cpu")），并在部署前于目标硬件上测试推理。

评论

支持 Markdown