视觉流水线毕业项目:Capstone

视觉流水线毕业项目:Capstone

本节摘要:生产级视觉系统是一条用数据契约缝合起来的模型与规则链。零件都在本章前面学过了,本毕业项目把它们端到端串起来。你将设计一条「检测 + 分类 + 结构化 JSON + 服务层」的最小可行流水线:用 Pydantic 给每个模型边界装上类型契约、把 Mask R-CNN/YOLO 检测器与 ConvNeXt-Tiny 分类器接进同一个服务、给每个失败路径(空检测、越界框、过小裁剪、坏上传、模型加载失败)配具名错误码,最后用 FastAPI 上一个能接图片上传、返回带分类检测结果的端点。读完本节,你掌握的不是又一个模型,而是后续所有视觉产品都能往里塞零件的稳定骨架。

对应原课程:Phase 4 · Lesson 16 · vision-pipeline-capstone(原英文 phases/04-computer-vision/16-vision-pipeline-capstone/docs/en.md)。本章毕业项目。

学习目标

阅读完本节,你应当能够:

1. 设计一条生产级视觉流水线——检测物体、分类、输出结构化 JSON,且每个失败路径都有处理。
2. 把检测器(Mask R-CNN 或 YOLO)、分类器(ConvNeXt-Tiny)和数据契约(Pydantic)接到同一个服务里。
3. 对端到端流水线做基准测试,找出第一个瓶颈(通常是预处理,然后是检测器)。
4. 上一个最小 FastAPI 服务:接受图片上传、跑流水线、返回带分类的检测结果。

一、问题与直觉

单个视觉模型有用;视觉产品是一串模型。货架巡检 = 检测器 + 商品分类器 + 价格 OCR;自动驾驶 = 2D 检测器 + 3D 检测器 + 分割器 + 跟踪器 + 规划器;医疗预筛 = 分割器 + 区域分类器 + 临床 UI。

把这些链串起来,正是区分 ML 原型与产品的那部分。模型之间的每个接口都是新 bug 的温床——每次坐标变换、每次归一化、每次掩码缩放,都是静默失败的候选。一条流水线的强度,取决于它最弱的那个接口。

本毕业项目搭最小可行流水线:检测 + 分类 + 结构化输出 + 服务层。Phase 4 其余一切都能塞进这套骨架:把 Mask R-CNN 换成 YOLOv8、加个 OCR 头、加个分割分支、加个跟踪器。架构稳定,零件可插拔。

流水线全貌

七阶段。两个模型阶段贵;其余五个阶段才是 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、转换色彩空间、缩放——这些都是 CPU 密集,且容易被忘。
2. 检测器占大头 GPU 时间。70~90% 的 GPU 时间花在检测前向上。
3. 后处理(NMS、RLE 编解码)GPU 上便宜,CPU 上贵。永远在真实目标上剖析。

知道分布,才能把优化变成一张排好优先级的清单。

失败模式

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

生产流水线要处理每一种,而不是写一个把失败藏起来的通用 try/except。每个失败都要有具名码和响应。

批处理

生产服务要服务多个客户端。跨请求批处理检测和分类能让吞吐翻几倍。代价是:等批填满带来的额外延迟。典型设置:最多攒 20ms 的请求,批在一起,处理,分发响应。torchserve 和 triton 原生支持;负载可预测的小服务自己撸个微批处理器。

二、从零实现

步骤 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、检测 300500 ms、分类 2040 ms、总计 350550 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:一个提示词——审查视觉服务代码的契约/响应形状违规,指出第一个会崩的 bug。
• skill-pipeline-budget-planner.md:一个技能——给定目标延迟和吞吐,给每个流水线阶段分配时间预算,并标记哪个阶段会先超预算。

五、练习

1. (简单) 在任意开放数据集的 10 张图上跑流水线。报告每阶段平均时间和每图检测数分布。
2. (中等) 给 Detection 加掩码输出字段并编码为 RLE。验证即便 10 个物体的图,JSON 也不超过 1 MB。
3. (困难) 在分类器前加微批处理器:最多攒 10 ms 的裁剪、一次 GPU 调用全分类、按请求返回结果。测量 5 并发请求/秒时的吞吐增益和新增延迟。

本节要点回顾

1. 视觉产品 = 模型链——货架巡检、自动驾驶、医疗预筛都是检测+分类+OCR+跟踪的串联。
2. 接口是 bug 温床——每次坐标变换、归一化、掩码缩放都可能静默失败;流水线强度取决于最弱接口。
3. 七阶段流水线——两个模型阶段贵,其余五阶段(预处理、裁剪、聚合、契约、响应)是 bug 藏身处。
4. Pydantic 数据契约——每个模型边界类型化,把静默失败变显式报错;五秒代码省一小时调试。
5. 延迟三真相——预处理常是最大 CPU 块、检测占 70~90% GPU 时间、后处理 GPU 便宜 CPU 贵。
6. 五类失败要显式处理——空检测、越界框、过小裁剪、坏上传(400)、模型加载失败(启动时)。
7. 微批处理换吞吐——攒 20 ms 请求批处理,torchserve/triton 原生支持。
8. 生产模板加四件——模型版本+权重哈希、每请求 trace ID、降级路径、安全过滤;torchserve/Triton/BentoML 开箱。
9. 骨架稳定、零件可插拔——后续可换 YOLOv8、加 OCR/分割/跟踪,架构不变。

本章前半部分到此告一段落。从下一节起,我们进入更前沿的专题:自监督学习、CLIP 开放词表、OCR 文档理解、图像检索、姿态估计、3D 高斯泼溅、扩散 Transformer、SAM3、视觉语言模型、单目深度、多目标跟踪、世界模型——把这些零件接进今天这条骨架,你就拥有了 2026 年生产级视觉系统的完整工具箱。