搭一条完整的视觉流水线 —— 综合项目

一个生产级视觉系统是一串模型和规则，用数据契约缝合起来。这些部件本阶段已经齐了；综合项目把它们端到端接起来。

类型： Build 语言： Python 前置要求： 阶段 4 第 01-15 课 预计时间： ~120 分钟

学习目标

• 设计一条生产视觉流水线，检测物体、对它们分类、产出结构化 JSON——每条失败路径都处理掉
• 把一个检测器（Mask R-CNN 或 YOLO）、一个分类器（ConvNeXt-Tiny）和一份数据契约（Pydantic）插进一个服务里
• 给端到端流水线做基准测试，找出第一个瓶颈（通常是预处理，然后是检测器）
• 交付一个极简 FastAPI 服务，接受图像上传、跑流水线、返回带分类的检测结果

问题背景

单个视觉模型有用；视觉产品是它们的链条。零售货架审计是检测器加产品分类器加价格 OCR 流水线。自动驾驶是 2D 检测器加 3D 检测器加分割器加追踪器加规划器。医学预筛是分割器加区域分类器加临床医生 UI。

把这些链条接起来，正是区分"ML 原型"和"产品"的那部分。模型之间每个接口都是 bug 的新藏身处。每个坐标变换、每次归一化、每次掩码 resize，都是无声故障的候选。一条流水线的强度等于它最弱的那个接口。

这个综合项目搭起最小可用流水线：检测 + 分类 + 结构化输出 + 一个服务层。Phase 4 里其余一切都嵌进这个骨架：把 Mask R-CNN 换成 YOLOv8、加个 OCR 头、加个分割分支、加个追踪器。架构稳定；部件可插拔。

核心概念

流水线

flowchart LR
    REQ["HTTP 请求<br/>+ 图像字节"] --> LOAD["解码<br/>+ 预处理"]
    LOAD --> DET["检测器<br/>(YOLO / Mask R-CNN)"]
    DET --> CROP["裁剪 + resize<br/>每个检测结果"]
    CROP --> CLS["分类器<br/>(ConvNeXt-Tiny)"]
    CLS --> AGG["聚合<br/>检测 + 类别"]
    AGG --> SCHEMA["Pydantic<br/>校验"]
    SCHEMA --> RESP["JSON 响应"]

    REQ -.->|错误| RESP

    style DET fill:#fef3c7,stroke:#d97706
    style CLS fill:#dbeafe,stroke:#2563eb
    style SCHEMA fill:#dcfce7,stroke:#16a34a

七个阶段。两个模型阶段很贵；其余五个阶段才是 bug 藏身的地方。

用 Pydantic 做数据契约

每个模型边界都变成一个带类型的对象。这把无声故障变成了响亮的故障。

Detection(
    box: tuple[float, float, float, float],   # (x1, y1, x2, y2)，绝对像素
    score: float,                              # [0, 1]
    class_id: int,                             # 来自检测器的标签映射
    mask: Optional[list[list[int]]],           # 若存在则 RLE 编码
)

PipelineResult(
    image_id: str,
    detections: list[Detection],
    classifications: list[Classification],
    inference_ms: float,
)

当一个检测器返回的框是 (cx, cy, w, h) 而不是 (x1, y1, x2, y2) 时，Pydantic 的校验在边界处失败，你立刻就发现了，而不是去调试一个悄悄返回空区域的下游裁剪。

延迟都去哪了

几乎每条视觉流水线里都成立的三个事实：

1. 预处理常常是最大的单个块。 解码 JPEG、转换色彩空间、resize——这些是 CPU 密集的，又容易被忘掉。
2. 检测器主导 GPU 时间。 70-90% 的 GPU 时间在检测的前向里。
3. 后处理（NMS、RLE 编解码）在 GPU 上便宜，在 CPU 上贵。 永远用真实的目标来 profile。

知道这个分布，就能把优化变成一份按优先级排好的清单。

失败模式

• 空检测 —— 返回空列表，别崩。记日志。
• 越界的框 —— 裁剪前 clamp 到图像尺寸。
• 极小的裁剪 —— 对小于分类器最小输入的框跳过分类。
• 损坏的上传 —— 返回 400 加一个具体错误码，不是 500。
• 模型加载失败 —— 在服务启动时失败，不是在第一个请求时。

一条生产流水线处理这些每一个，而不去写隐藏故障的通用 try/except。每个故障都有一个命名的码和一个响应。

批处理

一个生产服务服务多个客户端。跨请求批处理检测和分类能成倍提升吞吐。代价：等一个 batch 填满带来的额外延迟。典型做法：收集请求最多 20ms，凑成一批，处理，分发响应。torchserve 和 triton 原生这么做；负载可预测的小服务自己写个微批处理器。

v4-vision-pipeline

动手构建

第 1 步：数据契约

from pydantic import BaseModel, Field
from typing import List, Optional, Tuple

class Detection(BaseModel):
    box: Tuple[float, float, float, float]
    score: float = Field(ge=0, le=1)
    class_id: int = Field(ge=0)
    mask_rle: Optional[str] = None

class Classification(BaseModel):
    detection_index: int
    class_id: int
    class_name: str
    score: float = Field(ge=0, le=1)

class PipelineResult(BaseModel):
    image_id: str
    detections: List[Detection]
    classifications: List[Classification]
    inference_ms: float

五秒钟的代码，在任何正经流水线上省掉一小时的调试。

第 2 步：一个极简的 Pipeline 类

import time
import numpy as np
import torch
from PIL import Image

class VisionPipeline:
    def __init__(self, detector, classifier, class_names,
                 device="cpu", min_crop=32):
        self.detector = detector.to(device).eval()
        self.classifier = classifier.to(device).eval()
        self.class_names = class_names
        self.device = device
        self.min_crop = min_crop

    def preprocess(self, image):
        """
        image: PIL.Image 或 np.ndarray (H, W, 3) uint8
        返回: 设备上的 CHW float 张量
        """
        if isinstance(image, Image.Image):
            image = np.asarray(image.convert("RGB"))
        tensor = torch.from_numpy(image).permute(2, 0, 1).float() / 255.0
        return tensor.to(self.device)

    @torch.no_grad()
    def detect(self, image_tensor):
        return self.detector([image_tensor])[0]

    @torch.no_grad()
    def classify(self, crops):
        if len(crops) == 0:
            return []
        batch = torch.stack(crops).to(self.device)
        logits = self.classifier(batch)
        probs = logits.softmax(-1)
        scores, cls = probs.max(-1)
        return list(zip(cls.tolist(), scores.tolist()))

    def run(self, image, image_id="anonymous"):
        t0 = time.perf_counter()
        tensor = self.preprocess(image)
        det = self.detect(tensor)

        crops = []
        detections = []
        valid_indices = []
        for i, (box, score, cls) in enumerate(zip(det["boxes"], det["scores"], det["labels"])):
            x1, y1, x2, y2 = [max(0, int(b)) for b in box.tolist()]
            x2 = min(x2, tensor.shape[-1])
            y2 = min(y2, tensor.shape[-2])
            detections.append(Detection(
                box=(x1, y1, x2, y2),
                score=float(score),
                class_id=int(cls),
            ))
            if (x2 - x1) < self.min_crop or (y2 - y1) < self.min_crop:
                continue
            crop = tensor[:, y1:y2, x1:x2]
            crop = torch.nn.functional.interpolate(
                crop.unsqueeze(0),
                size=(224, 224),
                mode="bilinear",
                align_corners=False,
            )[0]
            crops.append(crop)
            valid_indices.append(i)

        class_preds = self.classify(crops)

        classifications = []
        for valid_idx, (cls_id, cls_score) in zip(valid_indices, class_preds):
            classifications.append(Classification(
                detection_index=valid_idx,
                class_id=int(cls_id),
                class_name=self.class_names[cls_id],
                score=float(cls_score),
            ))

        return PipelineResult(
            image_id=image_id,
            detections=detections,
            classifications=classifications,
            inference_ms=(time.perf_counter() - t0) * 1000,
        )

每个接口都带类型。每条失败路径都有一个具体的处理决策。

第 3 步：接一个检测器和一个分类器

from torchvision.models.detection import maskrcnn_resnet50_fpn_v2
from torchvision.models import convnext_tiny

# 用 ImageNet 预训练权重，不训练就得到一条真实感的流水线
detector = maskrcnn_resnet50_fpn_v2(weights="DEFAULT")
classifier = convnext_tiny(weights="DEFAULT")
class_names = [f"imagenet_class_{i}" for i in range(1000)]

pipe = VisionPipeline(detector, classifier, class_names)

# 用一张合成图像做冒烟测试
test_image = (np.random.rand(400, 600, 3) * 255).astype(np.uint8)
result = pipe.run(test_image, image_id="demo")
print(result.model_dump_json(indent=2)[:500])

第 4 步：FastAPI 服务

from fastapi import FastAPI, UploadFile, HTTPException
from io import BytesIO

app = FastAPI()
pipe = None  # 启动时初始化

@app.on_event("startup")
def load():
    global pipe
    detector = maskrcnn_resnet50_fpn_v2(weights="DEFAULT").eval()
    classifier = convnext_tiny(weights="DEFAULT").eval()
    pipe = VisionPipeline(detector, classifier, class_names=[f"c{i}" for i in range(1000)])

@app.post("/detect")
async def detect_endpoint(file: UploadFile):
    if file.content_type not in {"image/jpeg", "image/png", "image/webp"}:
        raise HTTPException(status_code=400, detail="unsupported image type")
    data = await file.read()
    try:
        img = Image.open(BytesIO(data)).convert("RGB")
    except Exception:
        raise HTTPException(status_code=400, detail="cannot decode image")
    result = pipe.run(img, image_id=file.filename or "upload")
    return result.model_dump()

用 uvicorn main:app --host 0.0.0.0 --port 8000 运行。用 curl -F 'file=@dog.jpg' http://localhost:8000/detect 测试。

第 5 步：给流水线做基准测试

import time

def benchmark(pipe, num_runs=20, image_size=(400, 600)):
    img = (np.random.rand(*image_size, 3) * 255).astype(np.uint8)
    pipe.run(img)  # 预热

    stages = {"preprocess": [], "detect": [], "classify": [], "total": []}
    for _ in range(num_runs):
        t0 = time.perf_counter()
        tensor = pipe.preprocess(img)
        t1 = time.perf_counter()
        det = pipe.detect(tensor)
        t2 = time.perf_counter()
        crops = []
        for box in det["boxes"]:
            x1, y1, x2, y2 = [max(0, int(b)) for b in box.tolist()]
            x2 = min(x2, tensor.shape[-1])
            y2 = min(y2, tensor.shape[-2])
            if (x2 - x1) >= pipe.min_crop and (y2 - y1) >= pipe.min_crop:
                crop = tensor[:, y1:y2, x1:x2]
                crop = torch.nn.functional.interpolate(
                    crop.unsqueeze(0), size=(224, 224), mode="bilinear", align_corners=False
                )[0]
                crops.append(crop)
        pipe.classify(crops)
        t3 = time.perf_counter()
        stages["preprocess"].append((t1 - t0) * 1000)
        stages["detect"].append((t2 - t1) * 1000)
        stages["classify"].append((t3 - t2) * 1000)
        stages["total"].append((t3 - t0) * 1000)

    for stage, times in stages.items():
        times.sort()
        print(f"{stage:12s}  p50={times[len(times)//2]:7.1f} ms  p95={times[int(len(times)*0.95)]:7.1f} ms")

CPU 上的典型输出：预处理约 3 ms，检测 300-500 ms，分类 20-40 ms，总计 350-550 ms。在 GPU 上，检测是 20-40 ms，预处理 + 分类相对而言开始更要紧。

实际使用

生产模板都收敛到同一个结构，外加：

• 模型版本化 —— 总是把模型名和权重哈希记进响应。
• 每请求的 trace ID —— 把每个请求每个阶段的耗时都记下来，好把慢响应和阶段关联起来。
• 降级路径 —— 如果分类器超时，返回不带分类的检测结果，而不是让整个请求失败。
• 安全过滤器 —— NSFW / PII 过滤器在分类后、响应离开服务前运行。
• 批量端点 —— 一个 /detect_batch，接受一个图像 URL 列表做批量处理。

生产服务用 torchserve、Triton Inference Server 和 BentoML，它们开箱即用地处理批处理、版本化、指标和健康检查。直接跑 FastAPI 对原型和小规模产品没问题。

拿去用

这一课产出：

• outputs/prompt-vision-service-shape-reviewer.md —— 一个 prompt，审查视觉服务代码里的契约/响应形状违规，点名第一个会崩的 bug。
• outputs/skill-pipeline-budget-planner.md —— 一个 skill，给定目标延迟和吞吐，给每个流水线阶段分配时间预算，并标出哪个阶段会最先超预算。

练习

1. （简单） 在任意开放数据集的 10 张图像上跑这条流水线。报告每个阶段的平均耗时和每张图检测数量的分布。
2. （中等） 给 Detection 加一个掩码输出字段，编码成 RLE。验证即便是 10 个物体的图像，JSON 也保持在 1MB 以下。
3. （困难） 在分类器前面加一个微批处理器：收集裁剪最多 10 ms，在一次 GPU 调用里全部分类，按请求返回结果。测量在每秒 5 个并发请求下的吞吐提升和增加的延迟。

关键术语

术语	大家嘴上怎么说	它实际是什么
流水线	"系统"	预处理、推理、后处理步骤的有序链条，每对之间有带类型的接口
数据契约	"schema"	每个阶段的输入输出都遵循的 Pydantic / dataclass 定义；在边界处抓集成 bug
预处理	"模型之前"	解码、色彩转换、resize、归一化；通常是最大的 CPU 时间消耗
后处理	"模型之后"	NMS、掩码 resize、阈值、RLE 编码；GPU 上便宜，CPU 上贵
微批处理器	"先收集再前向"	等一个固定窗口凑多个请求、跑一次批量前向的聚合器
Trace ID	"请求 id"	每请求一个标识符，在每个阶段记录，好把慢请求端到端追踪
失败码	"命名错误"	每个失败类一个具体错误码，而非通用 500；让客户端重试逻辑成为可能
健康检查	"就绪探针"	报告服务能否应答的廉价端点；负载均衡器依赖它

延伸阅读

• Full Stack Deep Learning — Deploying Models —— 生产 ML 部署的经典概览
• BentoML docs —— 带批处理、版本化和指标的服务框架
• torchserve docs —— PyTorch 官方服务库
• NVIDIA Triton Inference Server —— 带批处理和多模型支持的高吞吐服务