Skip to content
←

필사 모드: 用 BentoML 搭建 ML 模型服务流水线：从打包到 Kubernetes 部署

中文
0%
정확도 0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.

引言

训练一个 ML 模型和在生产环境中把它服务起来,是完全不同的两件事。BentoML 正是填补这道鸿沟的框架 — 它把模型打包成 API 服务,让你能在任何地方部署。相比直接用 Flask/FastAPI 手写 API,它提供了更体系化的方式。

BentoML 与手动实现的对比

项目Flask/FastAPI 手动实现BentoML
API 实现手动(路由、序列化)基于装饰器自动化
模型版本管理需要自行实现内置 Model Store
批处理需要自行实现内置 Adaptive Batching
Docker 构建手写 Dockerfile自动生成
GPU 支持手动配置声明式配置

安装与基本用法

pip install bentoml

保存模型

# save_model.py
import bentoml
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris

# 训练模型
X, y = load_iris(return_X_y=True)
model = RandomForestClassifier(n_estimators=100)
model.fit(X, y)

# 保存到 BentoML Model Store
saved_model = bentoml.sklearn.save_model(
    "iris_classifier",
    model,
    signatures={"predict": {"batchable": True}},
    labels={"owner": "ml-team", "stage": "production"},
    metadata={"accuracy": 0.96, "dataset": "iris"},
)

print(f"Model saved: {saved_model}")
# Model saved: Model(tag="iris_classifier:abc123")

# 查看已保存的模型
bentoml models list
# Tag                          Module    Size    Creation Time
# iris_classifier:abc123       sklearn   1.2MB   2026-03-03 05:00:00

定义 Service

# service.py
import bentoml
import numpy as np
from typing import Annotated

@bentoml.service(
    resources={"cpu": "2", "memory": "1Gi"},
    traffic={"timeout": 30, "concurrency": 32},
)
class IrisClassifier:
    model = bentoml.models.get("iris_classifier:latest")

    def __init__(self):
        self.clf = bentoml.sklearn.load_model(self.model)
        self.target_names = ["setosa", "versicolor", "virginica"]

    @bentoml.api
    def predict(
        self,
        features: Annotated[np.ndarray, bentoml.validators.Shape((4,))],
    ) -> dict:
        prediction = self.clf.predict([features])[0]
        probabilities = self.clf.predict_proba([features])[0]
        return {
            "class": self.target_names[prediction],
            "probability": float(max(probabilities)),
            "all_probabilities": {
                name: float(prob)
                for name, prob in zip(self.target_names, probabilities)
            },
        }

    @bentoml.api
    def predict_batch(
        self,
        features: Annotated[np.ndarray, bentoml.validators.Shape((-1, 4))],
    ) -> list[dict]:
        predictions = self.clf.predict(features)
        probabilities = self.clf.predict_proba(features)
        return [
            {
                "class": self.target_names[pred],
                "probability": float(max(probs)),
            }
            for pred, probs in zip(predictions, probabilities)
        ]

# 本地服务
bentoml serve service:IrisClassifier

# 测试
curl -X POST http://localhost:3000/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [5.1, 3.5, 1.4, 0.2]}'
# {"class": "setosa", "probability": 0.98, ...}

LLM 服务 — 集成 OpenLLM

# llm_service.py
import bentoml
from vllm import LLM, SamplingParams

@bentoml.service(
    resources={"gpu": 1, "gpu_type": "nvidia-a100"},
    traffic={"timeout": 120, "concurrency": 16},
)
class LLMService:
    def __init__(self):
        self.llm = LLM(
            model="meta-llama/Llama-3.1-8B-Instruct",
            tensor_parallel_size=1,
            max_model_len=8192,
            gpu_memory_utilization=0.9,
        )

    @bentoml.api
    async def generate(self, prompt: str, max_tokens: int = 512) -> str:
        sampling_params = SamplingParams(
            temperature=0.7,
            top_p=0.9,
            max_tokens=max_tokens,
        )
        outputs = self.llm.generate([prompt], sampling_params)
        return outputs[0].outputs[0].text

    @bentoml.api
    async def chat(self, messages: list[dict]) -> str:
        prompt = self._format_chat(messages)
        return await self.generate(prompt)

    def _format_chat(self, messages):
        formatted = ""
        for msg in messages:
            role = msg["role"]
            content = msg["content"]
            formatted += f"<|{role}|>\n{content}\n"
        formatted += "<|assistant|>\n"
        return formatted

多模型流水线

# pipeline_service.py
import bentoml
import numpy as np
from PIL import Image

@bentoml.service(resources={"cpu": "4", "memory": "4Gi"})
class ImageClassificationPipeline:
    # 组合多个模型
    preprocessor = bentoml.depends(ImagePreprocessor)
    classifier = bentoml.depends(ImageClassifier)
    postprocessor = bentoml.depends(ResultPostprocessor)

    @bentoml.api
    async def classify(self, image: Image.Image) -> dict:
        # 1. 预处理
        features = await self.preprocessor.process(image)

        # 2. 分类
        raw_result = await self.classifier.predict(features)

        # 3. 后处理
        result = await self.postprocessor.format(raw_result)

        return result

@bentoml.service(resources={"cpu": "1"})
class ImagePreprocessor:
    @bentoml.api
    async def process(self, image: Image.Image) -> np.ndarray:
        img = image.resize((224, 224))
        arr = np.array(img) / 255.0
        return arr.transpose(2, 0, 1)

@bentoml.service(resources={"gpu": 1})
class ImageClassifier:
    model = bentoml.models.get("resnet50:latest")

    def __init__(self):
        import torch
        self.model = bentoml.pytorch.load_model(self.model)
        self.model.eval()
        self.device = torch.device("cuda")
        self.model.to(self.device)

    @bentoml.api
    async def predict(self, features: np.ndarray) -> np.ndarray:
        import torch
        tensor = torch.tensor(features).unsqueeze(0).float().to(self.device)
        with torch.no_grad():
            output = self.model(tensor)
        return output.cpu().numpy()

Bento 构建与 Docker

bentofile.yaml

# bentofile.yaml
service: 'service:IrisClassifier'
labels:
  owner: ml-team
  project: iris-classifier
include:
  - '*.py'
python:
  packages:
    - scikit-learn==1.5.0
    - numpy
docker:
  python_version: '3.11'
  system_packages:
    - libgomp1
  env:
    BENTOML_PORT: '3000'

# 构建 Bento
bentoml build

# 查看构建好的 Bento
bentoml list
# Tag                              Size     Creation Time
# iris_classifier_service:xyz789   45MB     2026-03-03

# 生成 Docker 镜像
bentoml containerize iris_classifier_service:latest

# 用 Docker 运行
docker run -p 3000:3000 iris_classifier_service:latest

Kubernetes 部署

# k8s-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: iris-classifier
  namespace: ml-serving
spec:
  replicas: 3
  selector:
    matchLabels:
      app: iris-classifier
  template:
    metadata:
      labels:
        app: iris-classifier
    spec:
      containers:
        - name: bento
          image: registry.example.com/iris_classifier_service:latest
          ports:
            - containerPort: 3000
          resources:
            requests:
              cpu: '1'
              memory: '1Gi'
            limits:
              cpu: '2'
              memory: '2Gi'
          readinessProbe:
            httpGet:
              path: /healthz
              port: 3000
            initialDelaySeconds: 10
          livenessProbe:
            httpGet:
              path: /healthz
              port: 3000
            initialDelaySeconds: 30
---
apiVersion: v1
kind: Service
metadata:
  name: iris-classifier
  namespace: ml-serving
spec:
  selector:
    app: iris-classifier
  ports:
    - port: 80
      targetPort: 3000
  type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: iris-classifier
  namespace: ml-serving
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: iris-classifier
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 70

Adaptive Batching

BentoML 的核心功能之一,能自动把多个请求合并起来,把 GPU 利用率最大化。

@bentoml.service(
    traffic={
        "timeout": 30,
    },
)
class EmbeddingService:
    model = bentoml.models.get("sentence-transformer:latest")

    def __init__(self):
        from sentence_transformers import SentenceTransformer
        self.model = SentenceTransformer(self.model.path)

    @bentoml.api(
        batchable=True,
        batch_dim=0,
        max_batch_size=64,
        max_latency_ms=100,
    )
    async def encode(self, texts: list[str]) -> np.ndarray:
        # 单个请求会被自动合并成批次执行
        embeddings = self.model.encode(texts)
        return embeddings

监控

# 添加自定义指标
import bentoml
from prometheus_client import Counter, Histogram

prediction_counter = Counter(
    "predictions_total", "Total predictions", ["model", "class"]
)
latency_histogram = Histogram(
    "prediction_latency_seconds", "Prediction latency"
)

@bentoml.service
class MonitoredClassifier:
    @bentoml.api
    def predict(self, features: np.ndarray) -> dict:
        with latency_histogram.time():
            result = self.clf.predict([features])[0]
            prediction_counter.labels(
                model="iris_v1", class_name=result
            ).inc()
            return {"class": result}

# Prometheus 指标端点
curl http://localhost:3000/metrics

总结

BentoML 大幅降低了 ML 模型服务的复杂度:

  • 简单的 API 实现: 基于装饰器,几行代码就能生成 REST API
  • 模型版本管理: 内置 Model Store 做体系化管理
  • Adaptive Batching: 把 GPU 利用率最大化
  • Docker 自动化: 用 bentofile.yaml 实现可复现的构建
  • Kubernetes 原生: 配合 HPA 实现自动伸缩
测验：BentoML 理解度检查(7 题)

Q1. BentoML 的 Model Store 是什么？

一个把训练好的模型连同版本管理和元数据一起保存在本地的仓库。用 bentoml.sklearn.save_model() 等方法保存。

Q2. Adaptive Batching 的运作原理是什么？

自动收集单个请求,一旦达到 max_batch_size 或 max_latency_ms,就一次性处理,从而把 GPU 效率最大化。

Q3. bentoml.depends() 的作用是什么？

在多模型流水线中,把其他 BentoML 服务作为依赖注入进来,自动管理服务之间的通信。

Q4. bentofile.yaml 里定义的是什么？

声明服务入口点、Python 包依赖、Docker 配置、需要包含的文件等。

Q5. BentoML 的 /healthz 端点是做什么用的？

用于 Kubernetes 的 readiness/liveness probe,检查服务是否就绪、是否存活。

Q6. 如何指定 GPU 资源？

用 @bentoml.service(resources={"gpu": 1, "gpu_type": "nvidia-a100"}) 装饰器声明。

Q7. BentoML 相比直接用 Flask/FastAPI 实现有什么优势？

内置了模型版本管理、Adaptive Batching、Docker 自动构建、声明式资源管理等能力,能更快达到生产就绪。

현재 단락 (1/279)

训练一个 ML 模型和在生产环境中把它服务起来,是完全不同的两件事。**BentoML** 正是填补这道鸿沟的框架 — 它把模型打包成 API 服务,让你能在任何地方部署。相比直接用 Flask/Fa...

작성 글자: 0자원문 글자: 7,582자작성 단락: 0/279