KServe 模型服务完全指南：InferenceService、Canary 部署、Transformer、InferenceGraph 生产运维

Client Request
      │
      ▼
┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│ Transformer │────▶│  Predictor  │────▶│  Explainer  │
│ (前处理/    │     │ (模型推理)  │     │ (生成解释)  │
│  后处理)    │◀────│             │     │             │
└─────────────┘     └─────────────┘     └─────────────┘
      │
      ▼
Client Response

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: sklearn-iris
  namespace: ml-serving
  annotations:
    serving.kserve.io/deploymentMode: Serverless
spec:
  predictor:
    minReplicas: 1
    maxReplicas: 10
    scaleTarget: 5
    scaleMetric: concurrency
    model:
      modelFormat:
        name: sklearn
      storageUri: 's3://ml-models/sklearn/iris/v1'
      resources:
        requests:
          cpu: '500m'
          memory: '512Mi'
        limits:
          cpu: '1'
          memory: '1Gi'

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: llama3-vllm
  namespace: ml-serving
spec:
  predictor:
    minReplicas: 1
    maxReplicas: 4
    model:
      modelFormat:
        name: vLLM
      storageUri: 'pvc://llm-model-cache/Meta-Llama-3.1-8B-Instruct'
      args:
        - '--max-model-len=8192'
        - '--gpu-memory-utilization=0.90'
        - '--enable-chunked-prefill'
        - '--max-num-batched-tokens=16384'
        - '--tensor-parallel-size=2'
      resources:
        requests:
          cpu: '8'
          memory: '32Gi'
          nvidia.com/gpu: '2'
        limits:
          cpu: '16'
          memory: '64Gi'
          nvidia.com/gpu: '2'
    tolerations:
      - key: 'nvidia.com/gpu'
        operator: 'Exists'
        effect: 'NoSchedule'
    nodeSelector:
      gpu-type: 'a100'

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: fraud-detector
  namespace: ml-serving
  annotations:
    serving.kserve.io/deploymentMode: Serverless
spec:
  predictor:
    canaryTrafficPercent: 10
    minReplicas: 2
    maxReplicas: 20
    model:
      modelFormat:
        name: sklearn
      storageUri: 's3://ml-models/fraud/v2'
      resources:
        requests:
          cpu: '1'
          memory: '2Gi'
        limits:
          cpu: '2'
          memory: '4Gi'

#!/bin/bash
# canary-promote.sh - 渐进式 Canary 晋级脚本
ISVC_NAME="fraud-detector"
NAMESPACE="ml-serving"
STAGES=(10 30 50 80 100)
MAX_ERROR_RATE=0.05
OBSERVE_MINUTES=10

for pct in "${STAGES[@]}"; do
  echo "=== 将 Canary 流量更新为 ${pct}% ==="

  kubectl patch inferenceservice "$ISVC_NAME" -n "$NAMESPACE" \
    --type='json' \
    -p="[{\"op\": \"replace\", \"path\": \"/spec/predictor/canaryTrafficPercent\", \"value\": $pct}]"

  echo "正在观察 ${OBSERVE_MINUTES} 分钟的指标..."
  sleep $((OBSERVE_MINUTES * 60))

  # 从 Prometheus 查询 canary 错误率
  ERROR_RATE=$(curl -s "http://prometheus:9090/api/v1/query" \
    --data-urlencode "query=rate(revision_request_count{revision_name=~\".*canary.*\",response_code!=\"200\"}[5m]) / rate(revision_request_count{revision_name=~\".*canary.*\"}[5m])" \
    | jq -r '.data.result[0].value[1] // "0"')

  if (( $(echo "$ERROR_RATE > $MAX_ERROR_RATE" | bc -l) )); then
    echo "错误率 ${ERROR_RATE} 超过阈值 ${MAX_ERROR_RATE}。执行回滚。"
    kubectl patch inferenceservice "$ISVC_NAME" -n "$NAMESPACE" \
      --type='json' \
      -p='[{"op": "replace", "path": "/spec/predictor/canaryTrafficPercent", "value": 0}]'
    exit 1
  fi

  echo "错误率 ${ERROR_RATE} —— 处于正常范围。继续下一阶段。"
done

echo "=== Canary 晋级完成。v2 现已承载 100% 流量。 ==="

import kserve
from kserve import InferRequest, InferResponse, InferInput
from typing import Dict, List
import numpy as np
from PIL import Image
import requests
from io import BytesIO
import logging

logger = logging.getLogger(__name__)

class ImageTransformer(kserve.Model):
    """接收图像 URL,前处理后转发给 Predictor 的 Transformer"""

    def __init__(self, name: str, predictor_host: str):
        super().__init__(name)
        self.predictor_host = predictor_host
        self.target_size = (224, 224)
        self.mean = np.array([0.485, 0.456, 0.406])
        self.std = np.array([0.229, 0.224, 0.225])
        self.ready = False

    def load(self):
        """模型初始化。在此处放置 Warm-up 逻辑。"""
        logger.info("ImageTransformer 初始化完成")
        self.ready = True

    def preprocess(
        self, payload: Dict, headers: Dict = None
    ) -> InferRequest:
        """接收图像 URL,转换为归一化后的张量"""
        instances = payload.get("instances", [])
        processed_images = []

        for instance in instances:
            image_url = instance.get("image_url")
            response = requests.get(image_url, timeout=10)
            response.raise_for_status()

            image = Image.open(BytesIO(response.content)).convert("RGB")
            image = image.resize(self.target_size)

            # 转换为 numpy 数组后做归一化
            img_array = np.array(image, dtype=np.float32) / 255.0
            img_array = (img_array - self.mean) / self.std
            img_array = np.transpose(img_array, (2, 0, 1))  # CHW
            processed_images.append(img_array)

        input_tensor = np.stack(processed_images)

        infer_input = InferInput(
            name="input",
            shape=list(input_tensor.shape),
            datatype="FP32",
            data=input_tensor.tolist(),
        )
        return InferRequest(
            model_name=self.name,
            infer_inputs=[infer_input],
        )

    def postprocess(
        self, response: InferResponse, headers: Dict = None
    ) -> Dict:
        """将模型输出转换为易于阅读的格式"""
        predictions = response.outputs[0].data

        class_names = ["cat", "dog", "bird", "fish", "other"]
        results = []

        for pred in predictions:
            if isinstance(pred, list):
                probs = np.array(pred)
            else:
                probs = np.array([pred])

            top_idx = int(np.argmax(probs))
            results.append({
                "class": class_names[top_idx],
                "confidence": float(probs[top_idx]),
                "all_scores": {
                    name: float(score)
                    for name, score in zip(class_names, probs)
                },
            })

        return {"predictions": results}


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--predictor_host", required=True)
    parser.add_argument("--model_name", default="image-classifier")
    args = parser.parse_args()

    transformer = ImageTransformer(
        name=args.model_name,
        predictor_host=args.predictor_host,
    )
    transformer.load()
    kserve.ModelServer(workers=4).start([transformer])

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: image-classifier
  namespace: ml-serving
spec:
  transformer:
    minReplicas: 2
    maxReplicas: 15
    scaleTarget: 10
    scaleMetric: concurrency
    containers:
      - name: image-transformer
        image: registry.example.com/ml/image-transformer:v1.2.0
        args:
          - '--model_name=image-classifier'
        resources:
          requests:
            cpu: '1'
            memory: '2Gi'
          limits:
            cpu: '2'
            memory: '4Gi'
        env:
          - name: STORAGE_URI
            value: ''
  predictor:
    minReplicas: 1
    maxReplicas: 8
    model:
      modelFormat:
        name: pytorch
      storageUri: 's3://ml-models/image-classifier/resnet50-v2'
      resources:
        requests:
          cpu: '2'
          memory: '4Gi'
          nvidia.com/gpu: '1'
        limits:
          cpu: '4'
          memory: '8Gi'
          nvidia.com/gpu: '1'

apiVersion: serving.kserve.io/v1alpha1
kind: InferenceGraph
metadata:
  name: fraud-detection-ensemble
  namespace: ml-serving
  annotations:
    serving.kserve.io/propagateHeaders: 'x-request-id,x-trace-id'
spec:
  nodes:
    root:
      routerType: Sequence
      steps:
        - name: feature-enrichment
          serviceName: feature-enricher
          weight: 100
        - name: ensemble-node
          nodeName: model-ensemble
    model-ensemble:
      routerType: Ensemble
      steps:
        - name: xgboost-model
          serviceName: fraud-xgboost
          weight: 40
        - name: deep-model
          serviceName: fraud-deep-learning
          weight: 40
        - name: rule-engine
          serviceName: fraud-rule-engine
          weight: 20
    result-combiner:
      routerType: Sequence
      steps:
        - name: weighted-average
          serviceName: ensemble-combiner
          weight: 100

apiVersion: serving.kserve.io/v1alpha1
kind: InferenceGraph
metadata:
  name: recommendation-ab-test
  namespace: ml-serving
spec:
  nodes:
    root:
      routerType: Splitter
      steps:
        - name: model-v1-stable
          serviceName: recommender-v1
          weight: 80
        - name: model-v2-experiment
          serviceName: recommender-v2
          weight: 20

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: gpu-model
  namespace: ml-serving
  annotations:
    # 启用 Scale-to-Zero(默认值)
    serving.kserve.io/enable-scale-to-zero: 'true'
    # 最后一次请求后到 Scale-to-Zero 的等待时间
    autoscaling.knative.dev/scale-to-zero-pod-retention-period: '15m'
    # 缩容延迟(防止骤然缩容)
    autoscaling.knative.dev/scale-down-delay: '5m'
    # 可并发处理的请求数(GPU 模型应设置得较低)
    autoscaling.knative.dev/target: '2'
    autoscaling.knative.dev/metric: 'concurrency'
spec:
  predictor:
    minReplicas: 0
    maxReplicas: 4
    model:
      modelFormat:
        name: pytorch
      storageUri: 's3://ml-models/large-model/v1'
      resources:
        requests:
          nvidia.com/gpu: '1'
        limits:
          nvidia.com/gpu: '1'

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: llama3-vllm-scaler
  namespace: ml-serving
spec:
  scaleTargetRef:
    apiVersion: serving.kserve.io/v1beta1
    kind: InferenceService
    name: llama3-vllm
  minReplicaCount: 1
  maxReplicaCount: 8
  pollingInterval: 15
  cooldownPeriod: 300
  advanced:
    restoreToOriginalReplicaCount: true
    horizontalPodAutoscalerConfig:
      behavior:
        scaleUp:
          stabilizationWindowSeconds: 30
          policies:
            - type: Pods
              value: 2
              periodSeconds: 60
        scaleDown:
          stabilizationWindowSeconds: 300
          policies:
            - type: Pods
              value: 1
              periodSeconds: 120
  triggers:
    # 基于 vLLM 等待请求数的伸缩
    - type: prometheus
      metadata:
        serverAddress: http://prometheus.monitoring:9090
        metricName: vllm_waiting_requests
        query: |
          avg(vllm:num_requests_waiting{model_name="llama3-vllm"})
        threshold: '5'
    # 基于 KV Cache 使用率的伸缩
    - type: prometheus
      metadata:
        serverAddress: http://prometheus.monitoring:9090
        metricName: vllm_kv_cache_usage
        query: |
          avg(vllm:gpu_cache_usage_perc{model_name="llama3-vllm"})
        threshold: '0.85'

apiVersion: serving.kserve.io/v1alpha1
kind: LocalModelCache
metadata:
  name: llama3-cache
  namespace: ml-serving
spec:
  sourceModelUri: 's3://ml-models/Meta-Llama-3.1-8B-Instruct'
  nodeGroups:
    - name: a100-nodes
      nodeSelector:
        gpu-type: 'a100'
      persistentVolumeClaimSpec:
        accessModes:
          - ReadWriteOnce
        resources:
          requests:
            storage: 50Gi
        storageClassName: local-nvme

# 在前处理的最早阶段就限制图像尺寸
from PIL import Image

# PIL DecompressionBomb 防护设置
Image.MAX_IMAGE_PIXELS = 89_478_485  # 约 9500x9500

def safe_load_image(image_bytes: bytes, max_size: int = 2048):
    """内存安全的图像加载"""
    img = Image.open(BytesIO(image_bytes))

    # 若原始尺寸超过 max_size,立即缩放
    if max(img.size) > max_size:
        ratio = max_size / max(img.size)
        new_size = (int(img.width * ratio), int(img.height * ratio))
        img = img.resize(new_size, Image.LANCZOS)

    return img.convert("RGB")

spec:
  predictor:
    tolerations:
      - key: 'nvidia.com/gpu'
        operator: 'Exists'
        effect: 'NoSchedule'
    nodeSelector:
      gpu-type: 'a100'
    model:
      modelFormat:
        name: vLLM
      # ... 模型设置

apiVersion: v1
kind: ResourceQuota
metadata:
  name: ml-serving-quota
  namespace: ml-serving
spec:
  hard:
    requests.cpu: '64'
    requests.memory: '256Gi'
    requests.nvidia.com/gpu: '8'
    limits.cpu: '128'
    limits.memory: '512Gi'
    limits.nvidia.com/gpu: '8'
    persistentvolumeclaims: '20'