Полный конвейер компьютерного зрения — итоговый проект

Производственная система компьютерного зрения — это цепочка моделей и правил, соединённых контрактами данных. Все компоненты уже были в этой фазе; итоговый проект связывает их в сквозной процесс.

Тип: Соберите Языки: Python Предварительные требования: Фаза 4, уроки 01–15 Время: ~120 минут

Цели обучения

• Спроектировать производственный конвейер компьютерного зрения, который обнаруживает объекты, классифицирует их и выдаёт структурированный JSON — с обработкой каждого пути сбоя
• Объединить в одном сервисе детектор (Mask R-CNN или YOLO), классификатор (ConvNeXt-Tiny) и контракт данных (Pydantic)
• Измерить производительность сквозного конвейера и определить первое узкое место (обычно предобработку, затем детектор)
• Выпустить минимальный сервис FastAPI, принимающий загружаемое изображение, запускающий конвейер и возвращающий обнаружения с классификациями

Проблема

Отдельные визуальные модели полезны; визуальные продукты представляют собой их цепочки. Аудит полки в магазине — это детектор плюс классификатор товаров плюс OCR-конвейер для цен. Автономное вождение — это 2D-детектор плюс 3D-детектор плюс сегментатор плюс трекер плюс планировщик. Медицинский предварительный скрининг — это сегментатор плюс классификатор областей плюс интерфейс для клинициста.

Соединение этих цепочек — то, что отличает ML-прототип от продукта. Каждый интерфейс между моделями — новое место для ошибок. Каждое преобразование координат, каждая нормализация, каждое изменение размера маски — кандидат на тихий сбой. Конвейер настолько силён, насколько силён его слабейший интерфейс.

В этом итоговом проекте создаётся минимально жизнеспособный конвейер: обнаружение + классификация + структурированный вывод + слой обслуживания. Всё остальное из фазы 4 встраивается в этот каркас: замените Mask R-CNN на YOLOv8, добавьте OCR-голову, ветвь сегментации или трекер. Архитектура стабильна; компоненты взаимозаменяемы.

Концепция

Конвейер

Семь этапов. Два этапа с моделями дороги; на остальных пяти живут ошибки.

Контракты данных с Pydantic

Каждая граница модели становится типизированным объектом. Так тихие сбои становятся громкими.

Detection(
    box: tuple[float, float, float, float],   # (x1, y1, x2, y2), absolute pixels
    score: float,                              # [0, 1]
    class_id: int,                             # from detector's label map
    mask: Optional[list[list[int]]],           # RLE-encoded if present
)

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. Всегда профилируйте на реальной целевой платформе.

Знание распределения превращает оптимизацию в приоритизированный список.

Режимы отказа

• Пустые обнаружения — верните пустой список, не падайте. Запишите событие в журнал.
• Рамки за пределами изображения — ограничьте их размером изображения перед кадрированием.
• Слишком маленькие фрагменты — пропустите классификацию для рамок меньше минимального входа классификатора.
• Повреждённая загрузка — ответ 400 с конкретным кодом ошибки, а не 500.
• Сбой загрузки модели — завершите работу при запуске сервиса, а не при первом запросе.

Производственный конвейер обрабатывает каждый из этих случаев, не используя обобщённый try/except, скрывающий сбой. Для каждого сбоя нужны именованный код и ответ.

Пакетирование

Производственный сервис обслуживает нескольких клиентов. Пакетирование обнаружений и классификаций между запросами многократно повышает пропускную способность. Компромисс — дополнительная задержка в ожидании заполнения пакета. Типичная настройка: собирать запросы не более 20 мс, объединять их в пакет, обрабатывать и распределять ответы. 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 or np.ndarray (H, W, 3) uint8
        returns: CHW float tensor on device
        """
        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

# Use ImageNet-pretrained weights for a realistic pipeline without training
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)

# Smoke test with a synthetic image
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  # initialised on startup

@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)  # warm up

    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 мс, обнаружение 300–500 мс, классификация 20–40 мс, всего 350–550 мс. На GPU обнаружение занимает 20–40 мс, и предобработка + классификация начинают иметь большее относительное значение.

Используйте это

Производственные шаблоны сходятся к одной и той же структуре, а также включают:

• Версионирование моделей — всегда записывайте в ответ название модели и хеш весов.
• Идентификаторы трассировки на запрос — записывайте время каждого этапа для каждого запроса, чтобы можно было сопоставлять медленные ответы с этапами.
• Резервный путь — если классификатор превысил время ожидания, верните обнаружения без классификаций, а не завершайте весь запрос ошибкой.
• Фильтры безопасности — фильтры NSFW / PII запускаются после классификации, до того как ответ покинет сервис.
• Пакетный endpoint — /detect_batch, принимающий список URL изображений для массовой обработки.

Для производственного обслуживания torchserve, Triton Inference Server и BentoML из коробки обрабатывают пакетирование, версионирование, метрики и проверки работоспособности. Непосредственный запуск FastAPI подходит для прототипов и небольших продуктов.

Выпустите это

Этот урок создаёт:

• outputs/prompt-vision-service-shape-reviewer.md — промпт, проверяющий код сервиса компьютерного зрения на нарушения контрактов и формы ответов и называющий первую ломающую ошибку.
• outputs/skill-pipeline-budget-planner.md — навык, который по целевым задержке и пропускной способности распределяет бюджет времени между всеми этапами конвейера и отмечает этап, который первым не уложится в бюджет.

Упражнения

1. (Легко) Запустите конвейер на 10 изображениях из любого открытого набора данных. Сообщите среднее время на этап и распределение числа обнаружений на изображение.
2. (Средне) Добавьте в Detection поле вывода маски и кодируйте её как RLE. Убедитесь, что JSON остаётся меньше 1 МБ даже для изображения с 10 объектами.
3. (Сложно) Добавьте микропакетировщик перед классификатором: собирайте фрагменты до 10 мс, классифицируйте их все одним вызовом GPU и возвращайте результаты для каждого запроса. Измерьте прирост пропускной способности при 5 одновременных запросах в секунду и добавленную задержку.

Ключевые термины

Термин	Как говорят	Что это в действительности означает
Конвейер	«Система»	Упорядоченная цепочка шагов предобработки, инференса и постобработки с типизированным интерфейсом между каждой парой
Контракт данных	«Схема»	Определения Pydantic / dataclass, которым соответствует вход и выход каждого этапа; ловит ошибки интеграции на границе
Предобработка	«До модели»	Декодирование, преобразование цвета, изменение размера, нормализация; обычно крупнейший поглотитель времени CPU
Постобработка	«После модели»	NMS, изменение размера маски, порог, RLE-кодирование; дёшево на GPU, дорого на CPU
Микропакетировщик	«Собрать и отправить»	Агрегатор, ожидающий фиксированное окно несколько запросов и запускающий один пакетный прямой проход
Идентификатор трассировки	«Идентификатор запроса»	Идентификатор для каждого запроса, записываемый на каждом этапе, чтобы медленные запросы можно было отследить от начала до конца
Код сбоя	«Именованная ошибка»	Конкретный код ошибки для каждого класса сбоев вместо универсального 500; позволяет клиенту повторять запрос по правилам
Проверка работоспособности	«Проба готовности»	Дешёвый endpoint, сообщающий, может ли сервис отвечать; на него полагаются балансировщики нагрузки

Дополнительные материалы

• Full Stack Deep Learning — развёртывание моделей — канонический обзор развёртывания производственных ML-моделей
• Документация BentoML — фреймворк обслуживания с пакетированием, версионированием и метриками
• Документация torchserve — официальная библиотека PyTorch для обслуживания
• NVIDIA Triton Inference Server — высокопроизводительное обслуживание с пакетированием и несколькими моделями

---

Источник: Build a Complete Vision Pipeline — Capstone 04.15 — Компьютерное зрение в реальном времени — развёртывание на периферии · Фаза 04 — Компьютерное зрение · 04.17 — Самообучение представлений для зрения: SimCLR, DINO, MAE · Полный каталог