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

引言
BentoML 与手动实现的对比
安装与基本用法
- 保存模型
- 定义 Service
LLM 服务 — 集成 OpenLLM
多模型流水线
Bento 构建与 Docker
- bentofile.yaml
Kubernetes 部署
Adaptive Batching
监控
总结

引言

训练一个 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 自动构建、声明式资源管理等能力,能更快达到生产就绪。