mcpField
M
ml-api-endpoint

ml-api-endpoint

github.com/dengineproblem开发63 次安装
代码仓库

关于 ml-api-endpoint

面向机器学习工程师的 API 端点设计与部署技能，重点包括 FastAPI 推理接口、输入验证、性能优化、健康检查和 Docker 部署。

主要功能

  1. 01无状态 API 与响应格式设计
  2. 02FastAPI 推理端点示例
  3. 03输入验证与批量预测
  4. 04预测缓存与性能优化
  5. 05健康检查及 Docker 部署

适用场景

  1. 01为机器学习模型设计推理 API
  2. 02实现带输入校验的批量预测
  3. 03缓存确定性模型的预测结果
  4. 04为模型服务添加健康检查并部署

安装

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

$npx skills add https://github.com/dengineproblem/agents-monorepo --skill ml-api-endpoint

安装后在 Claude Code 中输入/ml-api-endpoint确认已加载。

详细介绍

ML API 端点专家

擅长设计和部署机器学习 API 端点。

核心原则

API 设计
  • 无状态设计：每个请求都包含所有必要信息
  • 统一的响应格式：规范成功和错误响应的结构
  • 版本策略：为模型更新做好规划
  • 输入验证：推理前进行严格验证

FastAPI 实现

基本 ML 端点
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
import joblib
import numpy as np

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

model = None

@app.on_event("startup")
async def load_model():
    global model
    model = joblib.load("model.pkl")

class PredictionInput(BaseModel):
    features: list[float]

    @validator('features')
    def validate_features(cls, v):
        if len(v) != 10:
            raise ValueError('Expected 10 features')
        return v

class PredictionResponse(BaseModel):
    prediction: float
    confidence: float | None = None
    model_version: str
    request_id: str

@app.post("/predict", response_model=PredictionResponse)
async def predict(input_data: PredictionInput):
    features = np.array([input_data.features])
    prediction = model.predict(features)[0]

    return PredictionResponse(
        prediction=float(prediction),
        model_version="v1",
        request_id=generate_request_id()
    )
批量预测
class BatchInput(BaseModel):
    instances: list[list[float]]

    @validator('instances')
    def validate_batch_size(cls, v):
        if len(v) > 100:
            raise ValueError('Batch size cannot exceed 100')
        return v

@app.post("/predict/batch")
async def batch_predict(input_data: BatchInput):
    features = np.array(input_data.instances)
    predictions = model.predict(features)

    return {
        "predictions": predictions.tolist(),
        "count": len(predictions)
    }

性能优化

模型缓存
class ModelCache:
    def __init__(self, ttl_seconds=300):
        self.cache = {}
        self.ttl = ttl_seconds

    def get(self, features):
        key = hashlib.md5(str(features).encode()).hexdigest()
        if key in self.cache:
            result, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return result
        return None

    def set(self, features, prediction):
        key = hashlib.md5(str(features).encode()).hexdigest()
        self.cache[key] = (prediction, time.time())

健康检查

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

@app.get("/metrics")
async def get_metrics():
    return {
        "requests_total": request_counter,
        "prediction_latency_avg": avg_latency,
        "error_rate": error_rate
    }

Docker 部署

FROM python:3.9-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", "--workers", "4"]

最佳实践

  • 对 I/O 操作使用 async/await
  • 验证数据类型、取值范围和业务规则
  • 为确定性模型缓存预测结果
  • 模型失败时使用备用响应进行处理
  • 记录预测结果、延迟和错误
  • 支持多个模型版本
  • 设置内存和 CPU 限制

评论

支持 Markdown